diff --git a/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadService.kt b/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadService.kt index 419eb7aa..487722a4 100644 --- a/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadService.kt +++ b/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadService.kt @@ -15,6 +15,7 @@ import android.net.NetworkRequest import android.os.Build import android.os.IBinder import android.os.PowerManager +import android.os.SystemClock import android.util.AtomicFile import androidx.core.app.NotificationCompat import gobackend.Gobackend @@ -56,6 +57,7 @@ class DownloadService : Service() { private const val VERIFICATION_REQUIRED_NOTIFICATION_ID = 4 private const val WAKELOCK_TAG = "SpotiFLAC:DownloadWakeLock" private const val WAKELOCK_RENEW_MS = 30 * 60 * 1000L + private const val WAKELOCK_RENEW_INTERVAL_MS = 15 * 60 * 1000L const val ACTION_START = "com.zarz.spotiflac.action.START_DOWNLOAD" const val ACTION_STOP = "com.zarz.spotiflac.action.STOP_DOWNLOAD" @@ -294,6 +296,9 @@ class DownloadService : Service() { if (progress.has("item_delta")) { state.put("item_delta", progress.get("item_delta")) } + if (progress.has("item_deltas")) { + state.put("item_deltas", progress.get("item_deltas")) + } state.put("snapshot_mode", "compact_with_delta") } } @@ -331,6 +336,7 @@ class DownloadService : Service() { private var nativeWorkerRequestChannel: Channel? = null @Volatile private var nativeWorkerPreparationComplete = true private var wakeLock: PowerManager.WakeLock? = null + @Volatile private var wakeLockLastRenewedAt = 0L private var currentTrackName = "" private var currentArtistName = "" internal var currentStatus = "preparing" @@ -338,6 +344,10 @@ class DownloadService : Service() { // Signature of the last home-screen widget push; keeps widget updates // event-driven (track/status/queue changes, 25% steps), never per byte. private var widgetSignature = "" + // NotificationManager updates are also deduplicated by the visible + // notification state. Progress writers can report many byte-level changes + // that render the same percentage/text. + private var lastNotificationSignature: String? = null internal var lastProgress = 0L internal var lastTotal = 0L internal var nativeWorkerRunId = "" @@ -346,6 +356,14 @@ class DownloadService : Service() { internal val nativeWorkerTerminalStatuses = mutableMapOf() internal val nativeReplayGainEntries = mutableListOf() internal val nativeReplayGainRequestAlbumKeys = mutableMapOf() + // One coordinator polls Go's delta progress stream for all native workers. + // Individual workers consume this cache instead of parsing the complete + // multi-progress JSON independently. + internal val nativeWorkerProgressLock = Any() + internal val nativeWorkerProgressItems = mutableMapOf() + internal var nativeWorkerProgressSeq = 0L + internal val nativeWorkerProgressEpoch = AtomicLong(0L) + @Volatile internal var nativeWorkerProgressJob: Job? = null internal val snapshotWriteLock = Any() internal val snapshotWriteSerial = AtomicLong(0L) internal var latestCommittedStateSnapshotSerial = 0L @@ -625,6 +643,7 @@ class DownloadService : Service() { private fun startForegroundService() { isRunning = true + lastNotificationSignature = null ensureWakeLock() @@ -1033,7 +1052,6 @@ class DownloadService : Service() { } if (nativeWorkerCancelRequested || generation != nativeWorkerGeneration) return - var progressJob: Job? = null var progressInitialized = false var retryCurrentRequest = false try { @@ -1073,31 +1091,6 @@ class DownloadService : Service() { ) Gobackend.initItemProgress(request.itemId) progressInitialized = true - progressJob = serviceScope.launch { - var lastSignature: String? = null - while (true) { - updateNativeWorkerItemProgress(request.itemId) - val signature = synchronized(nativeWorkerItems) { - nativeWorkerItems - .firstOrNull { it.itemId == request.itemId } - ?.let { - "${it.status}:${it.bytesReceived}:" + - "${it.bytesTotal}:${it.progress}" - } - } - if (signature != lastSignature) { - lastSignature = signature - writeNativeWorkerSnapshot( - isRunning = true, - isPaused = false, - currentItemId = request.itemId, - message = "Downloading", - settingsJson = settingsJson, - ) - } - delay(1000) - } - } currentStatus = "downloading" updateNativeWorkerItem(request.itemId) { it.status = "downloading" @@ -1107,8 +1100,6 @@ class DownloadService : Service() { Gobackend.downloadByStrategy(json) } } finally { - progressJob?.cancel() - progressJob = null updateNativeWorkerItemProgress(request.itemId) try { Gobackend.clearItemProgress(request.itemId) @@ -1319,7 +1310,6 @@ class DownloadService : Service() { includeItems = true, ) } finally { - progressJob?.cancel() if (progressInitialized) { updateNativeWorkerItemProgress(request.itemId) try { @@ -1348,6 +1338,7 @@ class DownloadService : Service() { val finalizerMutex = Mutex() val providerSemaphores = ConcurrentHashMap() val rateLimitAttempts = ConcurrentHashMap() + val progressCoordinatorJob = startNativeWorkerProgressCoordinator(generation) try { supervisorScope { @@ -1384,6 +1375,7 @@ class DownloadService : Service() { workers.joinAll() } } finally { + stopNativeWorkerProgressCoordinator(progressCoordinatorJob) if (generation == nativeWorkerGeneration) { cancelScheduledNativeWorkerItemsSnapshot() nativeWorkerRequestChannel = null @@ -1420,6 +1412,7 @@ class DownloadService : Service() { generation: Long ) { val rateLimitAttempts = mutableMapOf() + val progressCoordinatorJob = startNativeWorkerProgressCoordinator(generation) try { var requestIndex = 0 while (requestIndex < requests.size) { @@ -1467,41 +1460,11 @@ class DownloadService : Service() { includeItems = true ) - var progressJob: Job? = null try { Gobackend.initItemProgress(request.itemId) - progressJob = serviceScope.launch { - // The snapshot write is an AtomicFile open+fsync+ - // rename; skip ticks where progress hasn't moved. - var lastSignature: String? = null - while (true) { - updateNativeWorkerItemProgress(request.itemId) - val signature = synchronized(nativeWorkerItems) { - nativeWorkerItems - .firstOrNull { it.itemId == request.itemId } - ?.let { - "${it.status}:${it.bytesReceived}:" + - "${it.bytesTotal}:${it.progress}" - } - } - if (signature != lastSignature) { - lastSignature = signature - writeNativeWorkerSnapshot( - isRunning = true, - isPaused = false, - currentItemId = request.itemId, - message = "Downloading", - settingsJson = settingsJson - ) - } - delay(1000) - } - } val response = SafDownloadHandler.handle(this, request.requestJson) { json -> Gobackend.downloadByStrategy(json) } - progressJob.cancel() - progressJob = null if (generation != nativeWorkerGeneration) { // Superseded while blocked in the download call; the // new run owns the shared state now. @@ -1706,7 +1669,6 @@ class DownloadService : Service() { includeItems = true ) } finally { - progressJob?.cancel() updateNativeWorkerItemProgress(request.itemId) try { Gobackend.clearItemProgress(request.itemId) @@ -1721,6 +1683,7 @@ class DownloadService : Service() { } } } finally { + stopNativeWorkerProgressCoordinator(progressCoordinatorJob) if (generation == nativeWorkerGeneration) { cancelScheduledNativeWorkerItemsSnapshot() if (!nativeWorkerCancelRequested) { @@ -1750,10 +1713,15 @@ class DownloadService : Service() { } } + @Synchronized private fun ensureWakeLock() { val existingWakeLock = wakeLock if (existingWakeLock?.isHeld == true) { - existingWakeLock.acquire(WAKELOCK_RENEW_MS) + val now = SystemClock.elapsedRealtime() + if (now - wakeLockLastRenewedAt >= WAKELOCK_RENEW_INTERVAL_MS) { + existingWakeLock.acquire(WAKELOCK_RENEW_MS) + wakeLockLastRenewedAt = now + } return } if (existingWakeLock != null) { @@ -1768,12 +1736,14 @@ class DownloadService : Service() { setReferenceCounted(false) acquire(WAKELOCK_RENEW_MS) } + wakeLockLastRenewedAt = SystemClock.elapsedRealtime() } @Synchronized private fun releaseWakeLock() { val existingWakeLock = wakeLock wakeLock = null + wakeLockLastRenewedAt = 0L if (existingWakeLock?.isHeld == true) { try { existingWakeLock.release() @@ -1786,6 +1756,7 @@ class DownloadService : Service() { @Synchronized private fun stopForegroundService(cancelNativeWorker: Boolean = true) { cancelScheduledNativeWorkerItemsSnapshot() + cancelNativeWorkerProgressCoordinator() if (cancelNativeWorker) { nativeWorkerCancelRequested = true nativeWorkerPreparationComplete = true @@ -1811,6 +1782,7 @@ class DownloadService : Service() { nativeWorkerNetworkPaused = false nativeWorkerJob = null isRunning = false + lastNotificationSignature = null widgetSignature = "" try { DownloadQueueWidgetProvider.push(this, running = false) @@ -1828,17 +1800,49 @@ class DownloadService : Service() { return nativeWorkerItems.isNotEmpty() } } + + internal fun isNativeWorkerProgressActive(generation: Long): Boolean = + generation == nativeWorkerGeneration && + !nativeWorkerCancelRequested && + isRunning + + internal fun nativeWorkerCurrentItemIdSnapshot(): String = nativeWorkerCurrentItemId + @Synchronized internal fun updateNotification(progress: Long, total: Long) { if (!isRunning) return + // Keep the foreground-service wake lock alive even when the visible + // notification is unchanged and therefore deduplicated below. ensureWakeLock() + val visibleProgress = when { + total <= 0L -> "indeterminate" + total == NOTIFICATION_PERCENT_TOTAL -> + "percent:${(progress * 100 / total).toInt()}" + else -> { + // buildNotification renders one decimal place for MB and an + // integer percentage. Suppress byte-level updates that would + // produce the same visible notification. + val progressTenthsMb = progress / (1024L * 1024L / 10L) + val totalTenthsMb = total / (1024L * 1024L / 10L) + val percent = (progress * 100 / total).toInt() + "mb:${progressTenthsMb}:${totalTenthsMb}:$percent" + } + } + val signature = "$currentTrackName|$currentArtistName|$currentStatus|$queueCount|$visibleProgress" + if (signature == lastNotificationSignature) return + lastNotificationSignature = signature + val notification = buildNotification(progress, total) val manager = getSystemService(NotificationManager::class.java) manager.notify(NOTIFICATION_ID, notification) pushWidgetState(progress, total) } + internal fun maintainNativeWorkerWakeLock() { + if (isRunning) ensureWakeLock() + } + private fun pushWidgetState(progress: Long, total: Long) { val percent = if (total > 0) { ((progress * 100) / total).toInt().coerceIn(0, 100) diff --git a/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceSnapshot.kt b/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceSnapshot.kt index bf8add5b..ae04705d 100644 --- a/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceSnapshot.kt +++ b/android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceSnapshot.kt @@ -25,6 +25,7 @@ import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel import kotlinx.coroutines.delay +import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import org.json.JSONArray import org.json.JSONObject @@ -33,6 +34,18 @@ import java.util.concurrent.atomic.AtomicLong // Native-worker item state snapshots for the Flutter side. +/** + * The compact subset of Go progress needed by the Android worker UI. Keeping + * this as a Kotlin value avoids sharing mutable JSONObject instances between + * the polling coroutine and worker coroutines. + */ +internal data class NativeBackendProgress( + val status: String, + val bytesReceived: Long, + val bytesTotal: Long, + val progress: Double, +) + internal fun DownloadService.writeNativeWorkerSnapshot( isRunning: Boolean, isPaused: Boolean, @@ -41,10 +54,20 @@ internal fun DownloadService.writeNativeWorkerSnapshot( lastResult: JSONObject? = null, settingsJson: String = "", includeItems: Boolean = false, + progressItemIds: Collection? = null, + progressCoordinatorEpoch: Long? = null, snapshotSerial: Long = snapshotWriteSerial.incrementAndGet() ) { try { synchronized(snapshotWriteLock) { + // A stopped/superseded progress coordinator must not publish a + // newer running snapshot after the queue's terminal snapshot. + if ( + progressCoordinatorEpoch != null && + nativeWorkerProgressEpoch.get() != progressCoordinatorEpoch + ) { + return + } if (includeItems) { if (snapshotSerial < latestCommittedStateSnapshotSerial) return } else { @@ -67,13 +90,23 @@ internal fun DownloadService.writeNativeWorkerSnapshot( .put("snapshot_serial", snapshotSerial) .put("state_serial", if (includeItems) snapshotSerial else latestCommittedStateSnapshotSerial) .put("snapshot_mode", if (includeItems) "compact_items" else "delta") + if (includeItems) { + // The queue index is structural state. Progress deltas carry + // only the changed item; repeating every ID here made each + // tick grow linearly with a large queue. + snapshot.put("item_ids", nativeWorkerItemIds()) + } // Snapshot of the header before the per-item payload is // attached; served to pollers that already consumed this // items payload (see getNativeWorkerSnapshot). val headerCandidate = if (includeItems) snapshot.toString() else null - snapshot.put("item_ids", nativeWorkerItemIds()) if (includeItems) { snapshot.put("items", nativeWorkerItemsSnapshot(includeStatic = false)) + } else if (progressItemIds != null) { + snapshot.put( + "item_deltas", + nativeWorkerItemsSnapshot(progressItemIds, includeStatic = false), + ) } else { nativeWorkerItemSnapshot(currentItemId, includeStatic = false)?.let { snapshot.put("item_delta", it) @@ -209,60 +242,200 @@ internal fun DownloadService.updateNativeWorkerItem(itemId: String, updater: (Do } } -internal fun DownloadService.updateNativeWorkerItemProgress(itemId: String) { - try { - val raw = Gobackend.getAllDownloadProgress() +/** + * Polls the exported Go delta API once for the whole native queue. Workers + * update their item state from [nativeWorkerProgressItems] instead of making + * independent full-payload calls. The generation check prevents a delayed + * gomobile call from an old queue from contaminating a replacement queue. + */ +internal fun DownloadService.startNativeWorkerProgressCoordinator(generation: Long): Job { + nativeWorkerProgressJob?.cancel() + val coordinatorEpoch = nativeWorkerProgressEpoch.incrementAndGet() + synchronized(nativeWorkerProgressLock) { + nativeWorkerProgressItems.clear() + nativeWorkerProgressSeq = 0L + } + + val job = serviceScope.launch { + val lastSignatures = mutableMapOf() + while (isActive && isNativeWorkerProgressActive(generation)) { + maintainNativeWorkerWakeLock() + val changedItemIds = pollNativeWorkerProgress(generation) + val snapshotItemIds = mutableListOf() + for (itemId in changedItemIds) { + if (!updateNativeWorkerItemProgress(itemId, emitNotification = false)) { + continue + } + val signature = synchronized(nativeWorkerItems) { + nativeWorkerItems.firstOrNull { it.itemId == itemId }?.let { + "${it.status}:${it.bytesReceived}:${it.bytesTotal}:${it.progress}" + } + } + if (signature != null && lastSignatures[itemId] != signature) { + lastSignatures[itemId] = signature + snapshotItemIds.add(itemId) + } + } + + if (snapshotItemIds.isNotEmpty() && isNativeWorkerProgressActive(generation)) { + // Only the active item is visible in the foreground + // notification. Apply it once after all cache updates so a + // concurrent queue never emits one notification per worker. + val activeItemId = nativeWorkerCurrentItemIdSnapshot() + if (activeItemId.isNotBlank() && activeItemId in snapshotItemIds) { + updateNativeWorkerItemProgress(activeItemId, emitNotification = true) + } + val orderedItemIds = snapshotItemIds + .filter { it != activeItemId } + .let { ids -> + if (activeItemId in snapshotItemIds) ids + activeItemId else ids + } + // Commit every changed worker in one AtomicFile write. Writing + // one item_delta per worker would overwrite the same progress + // file repeatedly and expose only the last delta to Flutter. + writeNativeWorkerSnapshot( + isRunning = true, + isPaused = isNativeWorkerPaused(), + currentItemId = activeItemId.ifBlank { orderedItemIds.last() }, + message = if (isNativeWorkerPaused()) nativeWorkerPauseMessage() else "Downloading", + progressItemIds = orderedItemIds, + progressCoordinatorEpoch = coordinatorEpoch, + ) + } + delay(1000) + } + } + nativeWorkerProgressJob = job + return job +} + +internal fun DownloadService.stopNativeWorkerProgressCoordinator(job: Job? = nativeWorkerProgressJob) { + if (job == null) return + if (nativeWorkerProgressJob === job) { + nativeWorkerProgressJob = null + nativeWorkerProgressEpoch.incrementAndGet() + } + job.cancel() +} + +internal fun DownloadService.cancelNativeWorkerProgressCoordinator() { + stopNativeWorkerProgressCoordinator() + synchronized(nativeWorkerProgressLock) { + nativeWorkerProgressItems.clear() + nativeWorkerProgressSeq = 0L + } +} + +private fun DownloadService.pollNativeWorkerProgress(generation: Long): Set { + val sinceSeq = synchronized(nativeWorkerProgressLock) { nativeWorkerProgressSeq } + val raw = try { + Gobackend.getAllDownloadProgressDelta(sinceSeq) + } catch (_: Exception) { + return emptySet() + } + if (raw.isBlank() || !isNativeWorkerProgressActive(generation)) return emptySet() + + return try { val root = JSONObject(raw) - val items = root.optJSONObject("items") ?: return - val progress = items.optJSONObject(itemId) ?: return - val backendStatus = progress.optString("status", "downloading") - val bytesReceived = progress.optLong("bytes_received", 0L) - val bytesTotal = progress.optLong("bytes_total", 0L) + val nextSeq = root.optLong("seq", sinceSeq) + val reset = root.optBoolean("reset", false) + val updated = mutableMapOf() + root.optJSONObject("items")?.let { items -> + val keys = items.keys() + while (keys.hasNext()) { + val itemId = keys.next() + val item = items.optJSONObject(itemId) ?: continue + val bytesReceived = item.optLong("bytes_received", 0L).coerceAtLeast(0L) + val bytesTotal = item.optLong("bytes_total", 0L).coerceAtLeast(0L) + val progress = item.optDouble("progress", 0.0) + .takeUnless { it.isNaN() } + ?.coerceIn(0.0, 1.0) + ?: 0.0 + updated[itemId] = NativeBackendProgress( + status = item.optString("status", "downloading"), + bytesReceived = bytesReceived, + bytesTotal = bytesTotal, + progress = progress, + ) + } + } + val removed = mutableListOf() + root.optJSONArray("removed")?.let { ids -> + for (index in 0 until ids.length()) { + ids.optString(index, "").takeIf { it.isNotBlank() }?.let(removed::add) + } + } + synchronized(nativeWorkerProgressLock) { + if (!isNativeWorkerProgressActive(generation)) return emptySet() + if (reset) nativeWorkerProgressItems.clear() + nativeWorkerProgressItems.putAll(updated) + removed.forEach { nativeWorkerProgressItems.remove(it) } + if (nextSeq > nativeWorkerProgressSeq) { + nativeWorkerProgressSeq = nextSeq + } + } + updated.keys + } catch (_: Exception) { + emptySet() + } +} + +internal fun DownloadService.updateNativeWorkerItemProgress( + itemId: String, + emitNotification: Boolean = true, +): Boolean { + return try { + val progress = synchronized(nativeWorkerProgressLock) { + nativeWorkerProgressItems[itemId] + } ?: return false + + val backendStatus = progress.status if (backendStatus == "preparing") { - currentStatus = "preparing" updateNativeWorkerItem(itemId) { it.status = "preparing" it.progress = 0.0 it.bytesReceived = 0L it.bytesTotal = 0L } - lastProgress = 0L - lastTotal = 0L - updateNotification(0L, 0L) - return + if (emitNotification) { + currentStatus = "preparing" + lastProgress = 0L + lastTotal = 0L + updateNotification(0L, 0L) + } + return true } - val progressValue = if (bytesTotal > 0L) { - bytesReceived.toDouble() / bytesTotal.toDouble() + + val progressValue = if (progress.bytesTotal > 0L) { + progress.bytesReceived.toDouble() / progress.bytesTotal.toDouble() } else { - progress.optDouble("progress", 0.0) + progress.progress }.coerceIn(0.0, 1.0) - currentStatus = if (backendStatus == "finalizing") { - "finalizing" - } else { - "downloading" - } + val itemStatus = if (backendStatus == "finalizing") "finalizing" else "downloading" updateNativeWorkerItem(itemId) { - it.status = currentStatus + it.status = itemStatus it.progress = progressValue - it.bytesReceived = bytesReceived - it.bytesTotal = bytesTotal + it.bytesReceived = progress.bytesReceived + it.bytesTotal = progress.bytesTotal } - if (bytesTotal > 0L) { - lastProgress = bytesReceived - lastTotal = bytesTotal - updateNotification(bytesReceived, bytesTotal) - } else if (progressValue > 0.0) { - val percentProgress = (progressValue * DownloadService.NOTIFICATION_PERCENT_TOTAL).toLong() - .coerceIn(0L, DownloadService.NOTIFICATION_PERCENT_TOTAL) - lastProgress = percentProgress - lastTotal = DownloadService.NOTIFICATION_PERCENT_TOTAL - updateNotification(percentProgress, DownloadService.NOTIFICATION_PERCENT_TOTAL) - } else { - lastProgress = 0L - lastTotal = 0L - updateNotification(0L, 0L) + if (emitNotification) { + currentStatus = itemStatus + if (progress.bytesTotal > 0L) { + lastProgress = progress.bytesReceived + lastTotal = progress.bytesTotal + } else if (progressValue > 0.0) { + lastProgress = (progressValue * DownloadService.NOTIFICATION_PERCENT_TOTAL).toLong() + .coerceIn(0L, DownloadService.NOTIFICATION_PERCENT_TOTAL) + lastTotal = DownloadService.NOTIFICATION_PERCENT_TOTAL + } else { + lastProgress = 0L + lastTotal = 0L + } + updateNotification(lastProgress, lastTotal) } + true } catch (_: Exception) { + false } } @@ -324,6 +497,22 @@ internal fun DownloadService.nativeWorkerItemsSnapshot(includeStatic: Boolean): return array } +internal fun DownloadService.nativeWorkerItemsSnapshot( + itemIds: Collection, + includeStatic: Boolean, +): JSONArray { + val requested = itemIds.toSet() + val array = JSONArray() + synchronized(nativeWorkerItems) { + for (item in nativeWorkerItems) { + if (item.itemId in requested) { + array.put(nativeWorkerItemSnapshotLocked(item, includeStatic)) + } + } + } + return array +} + internal fun DownloadService.nativeWorkerItemSnapshotLocked(item: DownloadService.NativeWorkerItem, includeStatic: Boolean): JSONObject { val json = JSONObject() .put("item_id", item.itemId) diff --git a/android/app/src/main/kotlin/com/zarz/spotiflac/NativeDownloadFinalizer.kt b/android/app/src/main/kotlin/com/zarz/spotiflac/NativeDownloadFinalizer.kt index df7a98b9..98b01a30 100644 --- a/android/app/src/main/kotlin/com/zarz/spotiflac/NativeDownloadFinalizer.kt +++ b/android/app/src/main/kotlin/com/zarz/spotiflac/NativeDownloadFinalizer.kt @@ -40,6 +40,16 @@ object NativeDownloadFinalizer { // Native finalizer owns background-safe history writes while Flutter may be suspended. // Keep this schema contract in sync with Dart HistoryDatabase before bumping either side. const val HISTORY_SCHEMA_VERSION = 13 + // Keep one native connection for the process. Opening history.db and + // probing/migrating its schema for every finalized track was expensive, + // and a single guarded writer also prevents native finalizer calls from + // interleaving transactions on the same connection. Flutter/sqflite uses + // its own WAL connection, so the busy timeout remains configured once on + // this connection for cross-connection contention. + private val historyDatabaseLock = Any() + private var historyDatabase: SQLiteDatabase? = null + private var historyDatabasePath = "" + private var historyDatabaseSchemaVersion = 0 internal val activeFFmpegSessionIds = mutableSetOf() internal val nativeFFmpegSessionIds = BoundedRegistry(maxEntries = 256) internal val activeFFmpegSessionLock = Any() @@ -1346,26 +1356,18 @@ object NativeDownloadFinalizer { values: ContentValues, deduplicateTrack: Boolean = true, ) { - val dbFile = File(File(context.applicationInfo.dataDir, "app_flutter"), "history.db") - dbFile.parentFile?.mkdirs() - val db = SQLiteDatabase.openDatabase( - dbFile.absolutePath, - null, - SQLiteDatabase.OPEN_READWRITE or - SQLiteDatabase.CREATE_IF_NECESSARY or - SQLiteDatabase.ENABLE_WRITE_AHEAD_LOGGING, - ) - try { - configureHistoryDatabase(db) - db.beginTransaction() - try { - if (db.version > HISTORY_SCHEMA_VERSION) { - throw IllegalStateException( - "history schema v${db.version} is newer than native finalizer contract v$HISTORY_SCHEMA_VERSION" - ) - } - val needsBackfill = db.version < HISTORY_SCHEMA_VERSION - db.execSQL( + withHistoryDatabase(context) { db -> + val initializeSchema = historyDatabaseSchemaVersion != HISTORY_SCHEMA_VERSION + db.beginTransaction() + try { + if (initializeSchema) { + if (db.version > HISTORY_SCHEMA_VERSION) { + throw IllegalStateException( + "history schema v${db.version} is newer than native finalizer contract v$HISTORY_SCHEMA_VERSION" + ) + } + val needsBackfill = db.version < HISTORY_SCHEMA_VERSION + db.execSQL( """ CREATE TABLE IF NOT EXISTS history ( id TEXT PRIMARY KEY, @@ -1463,21 +1465,60 @@ object NativeDownloadFinalizer { db.execSQL("CREATE INDEX IF NOT EXISTS idx_history_queue_album ON history(sort_album, sort_track, id)") db.execSQL("CREATE INDEX IF NOT EXISTS idx_history_queue_genre ON history(sort_genre, sort_track, id)") db.execSQL("CREATE INDEX IF NOT EXISTS idx_history_queue_release ON history(sort_release, sort_track, id)") - if (db.version < HISTORY_SCHEMA_VERSION) db.version = HISTORY_SCHEMA_VERSION - if (deduplicateTrack) deleteDuplicateHistoryRows(db, values) - db.insertWithOnConflict("history", null, values, SQLiteDatabase.CONFLICT_REPLACE) - replaceHistoryPathKeys(db, values.getAsString("id"), values.getAsString("file_path")) - db.setTransactionSuccessful() - } finally { - db.endTransaction() + if (db.version < HISTORY_SCHEMA_VERSION) db.version = HISTORY_SCHEMA_VERSION + } + if (deduplicateTrack) deleteDuplicateHistoryRows(db, values) + db.insertWithOnConflict("history", null, values, SQLiteDatabase.CONFLICT_REPLACE) + replaceHistoryPathKeys(db, values.getAsString("id"), values.getAsString("file_path")) + db.setTransactionSuccessful() + } finally { + db.endTransaction() } - } finally { - db.close() + if (initializeSchema) { + historyDatabaseSchemaVersion = HISTORY_SCHEMA_VERSION + } + } + } + + private inline fun withHistoryDatabase( + context: Context, + block: (SQLiteDatabase) -> T, + ): T { + synchronized(historyDatabaseLock) { + val dbFile = File(File(context.applicationInfo.dataDir, "app_flutter"), "history.db") + dbFile.parentFile?.mkdirs() + val db = historyDatabase?.takeIf { + it.isOpen && historyDatabasePath == dbFile.absolutePath + } ?: run { + historyDatabase?.close() + val opened = SQLiteDatabase.openDatabase( + dbFile.absolutePath, + null, + SQLiteDatabase.OPEN_READWRITE or + SQLiteDatabase.CREATE_IF_NECESSARY or + SQLiteDatabase.ENABLE_WRITE_AHEAD_LOGGING, + ) + try { + configureHistoryDatabase(opened) + } catch (e: Exception) { + opened.close() + throw e + } + historyDatabase = opened + historyDatabasePath = dbFile.absolutePath + historyDatabaseSchemaVersion = 0 + opened + } + historyDatabase = db + return block(db) } } private fun configureHistoryDatabase(db: SQLiteDatabase) { runHistoryPragma(db, "PRAGMA busy_timeout = 5000", required = false) + // CONFLICT_REPLACE must fire the history delete trigger so the + // external-content FTS index does not retain the replaced rowid. + runHistoryPragma(db, "PRAGMA recursive_triggers = ON", required = false) runHistoryPragma(db, "PRAGMA synchronous = NORMAL", required = false) runHistoryPragma(db, "PRAGMA journal_mode = WAL", required = false) } diff --git a/go_backend/cancel.go b/go_backend/cancel.go index 06ec2460..1d4ad52b 100644 --- a/go_backend/cancel.go +++ b/go_backend/cancel.go @@ -90,6 +90,26 @@ func (r *cancelRegistry) requestCancel(id string) { r.mu.Unlock() } +// requestCancelActive marks every entry with live work as cancelled and +// returns its ID. Pending cancellation sentinels (refs <= 0) are deliberately +// ignored so a platform lifecycle callback cannot poison a future retry. +func (r *cancelRegistry) requestCancelActive() []string { + r.mu.Lock() + ids := make([]string, 0, len(r.entries)) + for id, entry := range r.entries { + if entry == nil || entry.refs <= 0 { + continue + } + entry.canceled = true + if entry.cancel != nil { + entry.cancel() + } + ids = append(ids, id) + } + r.mu.Unlock() + return ids +} + func (r *cancelRegistry) isCancelled(id string) bool { if id == "" { return false @@ -149,6 +169,14 @@ func cancelDownload(itemID string) { RemoveItemProgress(itemID) } +func cancelAllActiveDownloads() []string { + itemIDs := downloadCancels.requestCancelActive() + for _, itemID := range itemIDs { + RemoveItemProgress(itemID) + } + return itemIDs +} + func isDownloadCancelled(itemID string) bool { return downloadCancels.isCancelled(itemID) } diff --git a/go_backend/exports_download.go b/go_backend/exports_download.go index 9c6d2418..cea7a5c5 100644 --- a/go_backend/exports_download.go +++ b/go_backend/exports_download.go @@ -444,6 +444,18 @@ func CancelDownload(itemID string) { cancelDownload(itemID) } +// CancelAllActiveDownloads is a lifecycle safety valve for platforms that are +// about to suspend the process. It only cancels entries with live work and +// does not create cancellation flags for queued/future items. +func CancelAllActiveDownloads() string { + itemIDs := cancelAllActiveDownloads() + payload, err := json.Marshal(itemIDs) + if err != nil { + return "[]" + } + return string(payload) +} + // ResetDownloadCancel drops a pre-registered cancellation flag for an item // with no active download, so a user-initiated retry does not consume a stale // cancel and abort instantly. Entries with live references are left alone. diff --git a/go_backend/extension_lifecycle_signed_session_timeout_test.go b/go_backend/extension_lifecycle_signed_session_timeout_test.go new file mode 100644 index 00000000..75b4ac0e --- /dev/null +++ b/go_backend/extension_lifecycle_signed_session_timeout_test.go @@ -0,0 +1,270 @@ +package gobackend + +import ( + "context" + "errors" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/dop251/goja" +) + +func TestInitializeVMLockedBoundsTopLevelScript(t *testing.T) { + previousTimeout := extensionLifecycleTimeout + extensionLifecycleTimeout = 10 * time.Millisecond + t.Cleanup(func() { extensionLifecycleTimeout = previousTimeout }) + + sourceDir := t.TempDir() + indexPath := filepath.Join(sourceDir, "index.js") + if err := os.WriteFile(indexPath, []byte("registerExtension({}); while (true) {}"), 0600); err != nil { + t.Fatal(err) + } + ext := &loadedExtension{ + ID: "lifecycle-top-level-timeout", + Manifest: &ExtensionManifest{Name: "lifecycle-top-level-timeout"}, + SourceDir: sourceDir, + DataDir: t.TempDir(), + } + + err := initializeVMLocked(ext) + if err == nil || !IsTimeoutError(err) { + t.Fatalf("initialize error = %v, want timeout", err) + } + if ext.VM != nil || ext.runtime != nil || ext.initialized { + t.Fatalf("timed-out VM was not discarded: VM=%v runtime=%v initialized=%v", ext.VM, ext.runtime, ext.initialized) + } +} + +func TestInitializeAndCleanupLifecycleCallbacksAreBounded(t *testing.T) { + previousTimeout := extensionLifecycleTimeout + extensionLifecycleTimeout = 10 * time.Millisecond + t.Cleanup(func() { extensionLifecycleTimeout = previousTimeout }) + + vm := goja.New() + if _, err := vm.RunString(`extension = { + initialize: function() { while (true) {} }, + cleanup: function() { while (true) {} } + }`); err != nil { + t.Fatal(err) + } + + if err := initializeExtensionRuntimeWithSettings(vm, "lifecycle-callback-timeout", map[string]any{"quality": "lossless"}); err == nil || !IsTimeoutError(err) { + t.Fatalf("initialize callback error = %v, want timeout", err) + } + if err := runCleanupOnVM(vm); err == nil || !IsTimeoutError(err) { + t.Fatalf("cleanup callback error = %v, want timeout", err) + } +} + +func TestLifecycleTimeoutQuarantinesUnresponsiveCleanupVM(t *testing.T) { + previousTimeout := extensionLifecycleTimeout + previousGrace := jsInterruptGracePeriod + extensionLifecycleTimeout = 10 * time.Millisecond + jsInterruptGracePeriod = 10 * time.Millisecond + t.Cleanup(func() { + extensionLifecycleTimeout = previousTimeout + jsInterruptGracePeriod = previousGrace + }) + + vm := goja.New() + release := make(chan struct{}) + if err := vm.Set("block", func() { <-release }); err != nil { + t.Fatal(err) + } + if _, err := vm.RunString(`extension = { cleanup: function() { block(); } }`); err != nil { + t.Fatal(err) + } + ext := &loadedExtension{ + ID: "lifecycle-cleanup-quarantine", + Manifest: &ExtensionManifest{Name: "lifecycle-cleanup-quarantine"}, + VM: vm, + runtime: &extensionRuntime{}, + } + + done := make(chan error, 1) + go func() { + done <- func() error { ext.VMMu.Lock(); defer ext.VMMu.Unlock(); teardownVMLocked(ext); return nil }() + }() + + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("teardown did not return after cleanup timeout") + } + ext.VMMu.Lock() + vmRemaining, runtimeRemaining := ext.VM, ext.runtime + ext.VMMu.Unlock() + if vmRemaining != nil || runtimeRemaining != nil || !hasQuarantinedRuntime(ext) { + t.Fatalf("unsafe cleanup was not quarantined: VM=%v runtime=%v quarantined=%v", vmRemaining, runtimeRemaining, hasQuarantinedRuntime(ext)) + } + close(release) + deadline := time.Now().Add(time.Second) + for hasQuarantinedRuntime(ext) && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if hasQuarantinedRuntime(ext) { + t.Fatal("quarantined cleanup runtime did not finish") + } +} + +func TestSignedSessionGrantRetryHonorsCancellationAndReleasesCoordinator(t *testing.T) { + previousWait := signedSessionRetryWaitContext + previousLegacyWait := signedSessionRetryWait + signedSessionRetryWait = nil + signedSessionRetryWaitContext = func(ctx context.Context, _ time.Duration) error { + <-ctx.Done() + return ctx.Err() + } + t.Cleanup(func() { + signedSessionRetryWait = previousLegacyWait + signedSessionRetryWaitContext = previousWait + }) + + var calls atomic.Int32 + transport := roundTripFunc(func(req *http.Request) (*http.Response, error) { + calls.Add(1) + return &http.Response{ + StatusCode: http.StatusTooManyRequests, + Header: http.Header{"Retry-After": []string{"300"}}, + Body: io.NopCloser(strings.NewReader(`{}`)), + Request: req, + }, nil + }) + runtime := newSignedSessionTestRuntime(t, "signed-cancel", transport) + runtime.manifest.SignedSession = &SignedSessionConfig{ + Namespace: "signed-cancel", + BaseURL: "https://auth.example.com", + } + + ctx, cancel := context.WithCancel(context.Background()) + errCh := make(chan error, 1) + go func() { errCh <- runtime.exchangeSignedSessionGrantContext(ctx, "grant-cancel") }() + + deadline := time.Now().Add(time.Second) + for calls.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if calls.Load() == 0 { + t.Fatal("exchange request was not started") + } + + coordinator, err := runtime.signedSessionCoordinator(signedSessionConfigWithDefaults(runtime.manifest.SignedSession)) + if err != nil { + t.Fatal(err) + } + lockAcquired := make(chan struct{}) + go func() { + coordinator.mu.Lock() + coordinator.mu.Unlock() + close(lockAcquired) + }() + select { + case <-lockAcquired: + case <-time.After(time.Second): + t.Fatal("coordinator mutex remained held during Retry-After wait") + } + + cancel() + select { + case err := <-errCh: + if !errors.Is(err, context.Canceled) { + t.Fatalf("exchange error = %v, want context cancellation", err) + } + case <-time.After(time.Second): + t.Fatal("cancelled exchange did not return") + } + + coordinator.mu.Lock() + inFlight := coordinator.exchangeInFlight + coordinator.mu.Unlock() + if inFlight { + t.Fatal("coordinator exchange lease remained in flight after cancellation") + } +} + +func TestSignedSessionClearInvalidatesInFlightExchangeCommit(t *testing.T) { + requestStarted := make(chan struct{}) + releaseResponse := make(chan struct{}) + transport := roundTripFunc(func(req *http.Request) (*http.Response, error) { + close(requestStarted) + select { + case <-releaseResponse: + case <-req.Context().Done(): + return nil, req.Context().Err() + } + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader( + `{"session_id":"late","session_secret":"late-secret","expires_at":"2099-01-01T00:00:00Z"}`, + )), + Request: req, + }, nil + }) + runtime := newSignedSessionTestRuntime(t, "signed-clear-in-flight", transport) + runtime.manifest.SignedSession = &SignedSessionConfig{ + Namespace: "signed-clear-in-flight", + BaseURL: "https://auth.example.com", + } + + errCh := make(chan error, 1) + go func() { errCh <- runtime.exchangeSignedSessionGrant("grant-before-clear") }() + select { + case <-requestStarted: + case <-time.After(time.Second): + t.Fatal("exchange request was not started") + } + + clearResult := runtime.signedSessionClear(goja.FunctionCall{}).Export().(map[string]any) + if clearResult["success"] != true { + t.Fatalf("clear result = %#v, want success", clearResult) + } + close(releaseResponse) + select { + case err := <-errCh: + if err == nil || !strings.Contains(err.Error(), "superseded by session clear") { + t.Fatalf("exchange error = %v, want clear-generation rejection", err) + } + case <-time.After(time.Second): + t.Fatal("exchange did not finish") + } + + config := signedSessionConfigWithDefaults(runtime.manifest.SignedSession) + record, err := runtime.loadSignedSession(config) + if err != nil { + t.Fatal(err) + } + if record.SessionID != "" || record.SessionSecret != "" || record.ExpiresAt != "" { + t.Fatalf("cleared session was resurrected: %#v", record) + } +} + +func TestCancelAllActiveDownloadsDoesNotPoisonIdleItems(t *testing.T) { + activeContext, activeCancel := context.WithCancel(context.Background()) + idleContext, idleCancel := context.WithCancel(context.Background()) + t.Cleanup(idleCancel) + registry := &cancelRegistry{entries: map[string]*cancelEntry{ + "active": {ctx: activeContext, cancel: activeCancel, refs: 1}, + "idle": {ctx: idleContext, cancel: idleCancel, refs: 0}, + }} + + ids := registry.requestCancelActive() + if len(ids) != 1 || ids[0] != "active" { + t.Fatalf("cancelled IDs = %v, want [active]", ids) + } + if !errors.Is(activeContext.Err(), context.Canceled) { + t.Fatalf("active context error = %v, want cancellation", activeContext.Err()) + } + if idleContext.Err() != nil || registry.entries["idle"].canceled { + t.Fatal("idle entry was poisoned by active cancellation") + } +} diff --git a/go_backend/extension_manager.go b/go_backend/extension_manager.go index 23c1b08c..0e0bff8d 100644 --- a/go_backend/extension_manager.go +++ b/go_backend/extension_manager.go @@ -196,22 +196,28 @@ func (m *extensionManager) loadExtensionFromFileLocked(filePath string) (*loaded } m.mu.Lock() - defer m.mu.Unlock() - if _, exists := m.extensions[manifest.Name]; exists { + m.mu.Unlock() return nil, fmt.Errorf("extension '%s' was installed by another process", manifest.DisplayName) } - extDir, err := managedExtensionPath(m.extensionsDir, manifest.Name) + extensionsDir := m.extensionsDir + dataDir := m.dataDir + extDir, err := managedExtensionPath(extensionsDir, manifest.Name) if err != nil { + m.mu.Unlock() return nil, err } if _, err := os.Lstat(extDir); err == nil { + m.mu.Unlock() return nil, fmt.Errorf("extension directory already exists for %q", manifest.Name) } else if !os.IsNotExist(err) { + m.mu.Unlock() return nil, fmt.Errorf("failed to inspect extension directory: %w", err) } - stagingDir, err := os.MkdirTemp(m.extensionsDir, "."+manifest.Name+"-install-*") + m.mu.Unlock() + + stagingDir, err := os.MkdirTemp(extensionsDir, "."+manifest.Name+"-install-*") if err != nil { return nil, fmt.Errorf("failed to create extension staging directory: %w", err) } @@ -225,7 +231,7 @@ func (m *extensionManager) loadExtensionFromFileLocked(filePath string) (*loaded return nil, err } - extDataDir, err := managedExtensionPath(m.dataDir, manifest.Name) + extDataDir, err := managedExtensionPath(dataDir, manifest.Name) if err != nil { return nil, err } @@ -252,7 +258,15 @@ func (m *extensionManager) loadExtensionFromFileLocked(filePath string) (*loaded stagingCommitted = true ext.SourceDir = extDir + m.mu.Lock() + if _, exists := m.extensions[manifest.Name]; exists { + m.mu.Unlock() + teardownExtension(ext) + _ = os.RemoveAll(extDir) + return nil, fmt.Errorf("extension '%s' was installed by another process", manifest.DisplayName) + } m.extensions[manifest.Name] = ext + m.mu.Unlock() GoLog("[Extension] Loaded extension: %s v%s\n", manifest.DisplayName, manifest.Version) return ext, nil @@ -330,19 +344,23 @@ func teardownExtension(ext *loadedExtension) { func (m *extensionManager) UnloadExtension(extensionID string) error { m.mu.Lock() - defer m.mu.Unlock() - ext, exists := m.extensions[extensionID] if !exists { + m.mu.Unlock() return fmt.Errorf("extension not found") } ext.Enabled = false + // Remove the extension from the manager before running user cleanup. New + // operations can no longer acquire it, while existing operations keep their + // per-extension VMMu lease and are allowed to finish before teardown. + delete(m.extensions, extensionID) + m.mu.Unlock() + ext.VMMu.Lock() teardownVMLocked(ext) ext.VMMu.Unlock() - delete(m.extensions, extensionID) GoLog("[Extension] Unloaded extension: %s\n", extensionID) return nil @@ -372,24 +390,39 @@ func (m *extensionManager) GetAllExtensions() []*loadedExtension { func (m *extensionManager) SetExtensionEnabled(extensionID string, enabled bool) error { m.mu.Lock() - defer m.mu.Unlock() - ext, exists := m.extensions[extensionID] if !exists { + m.mu.Unlock() return fmt.Errorf("extension not found") } if enabled { ext.Enabled = true - if err := ext.ensureRuntimeReady(); err != nil { + } else { + ext.Enabled = false + ext.Error = "" + } + m.mu.Unlock() + + if enabled { + ext.VMMu.Lock() + if !m.isManagedExtensionEnabled(extensionID, ext) { + ext.VMMu.Unlock() + return fmt.Errorf("extension is no longer installed") + } + err := ensureRuntimeReadyLocked(ext, true) + ext.VMMu.Unlock() + if err != nil { + m.mu.Lock() + if m.extensions[extensionID] == ext { + ext.Enabled = false + } + m.mu.Unlock() store := GetExtensionSettingsStore() - ext.Enabled = false _ = store.Set(extensionID, "_enabled", false) return err } } else { - ext.Enabled = false - ext.Error = "" ext.VMMu.Lock() teardownVMLocked(ext) ext.VMMu.Unlock() @@ -404,6 +437,21 @@ func (m *extensionManager) SetExtensionEnabled(extensionID string, enabled bool) return nil } +// isManagedExtensionEnabled validates an operation lease after it has acquired +// the per-extension VM lock. The manager lock is deliberately not held while +// lifecycle/user JavaScript runs, so list/unload operations remain responsive. +func (m *extensionManager) isManagedExtensionEnabled(extensionID string, ext *loadedExtension) bool { + m.mu.RLock() + defer m.mu.RUnlock() + return m.extensions[extensionID] == ext && ext.Enabled +} + +func (m *extensionManager) isManagedExtension(extensionID string, ext *loadedExtension) bool { + m.mu.RLock() + defer m.mu.RUnlock() + return m.extensions[extensionID] == ext +} + func (m *extensionManager) LoadExtensionsFromDirectory(dirPath string) ([]string, []error) { var loaded []string var errors []error @@ -444,7 +492,12 @@ func (m *extensionManager) LoadExtensionsFromDirectory(dirPath string) ([]string func (m *extensionManager) loadExtensionFromDirectory(dirPath string) (*loadedExtension, error) { m.mu.Lock() - defer m.mu.Unlock() + locked := true + defer func() { + if locked { + m.mu.Unlock() + } + }() manifestPath := filepath.Join(dirPath, "manifest.json") manifestData, err := os.ReadFile(manifestPath) @@ -486,6 +539,8 @@ func (m *extensionManager) loadExtensionFromDirectory(dirPath string) (*loadedEx DataDir: extDataDir, SourceDir: dirPath, } + m.mu.Unlock() + locked = false store := GetExtensionSettingsStore() if enabledVal, err := store.Get(manifest.Name, "_enabled"); err == nil { @@ -501,7 +556,17 @@ func (m *extensionManager) loadExtensionFromDirectory(dirPath string) (*loadedEx GoLog("[Extension] Failed to validate extension %s: %v\n", manifest.Name, err) } + m.mu.Lock() + locked = true + if _, exists := m.extensions[manifest.Name]; exists { + m.mu.Unlock() + locked = false + teardownExtension(ext) + return nil, fmt.Errorf("extension '%s' was installed by another process", manifest.DisplayName) + } m.extensions[manifest.Name] = ext + m.mu.Unlock() + locked = false GoLog("[Extension] Loaded extension: %s v%s\n", manifest.DisplayName, manifest.Version) return ext, nil @@ -838,16 +903,18 @@ func (m *extensionManager) GetInstalledExtensionsJSON() (string, error) { } func (m *extensionManager) InitializeExtension(extensionID string, settings map[string]any) error { - m.mu.Lock() - defer m.mu.Unlock() - + m.mu.RLock() ext, exists := m.extensions[extensionID] + m.mu.RUnlock() if !exists { return fmt.Errorf("extension not found") } ext.VMMu.Lock() defer ext.VMMu.Unlock() + if !m.isManagedExtension(extensionID, ext) { + return fmt.Errorf("extension is no longer installed") + } if err := ensureRuntimeReadyLocked(ext, false); err != nil { return err @@ -856,20 +923,25 @@ func (m *extensionManager) InitializeExtension(extensionID string, settings map[ } func (m *extensionManager) CleanupExtension(extensionID string) error { - m.mu.Lock() - defer m.mu.Unlock() - + m.mu.RLock() ext, exists := m.extensions[extensionID] + m.mu.RUnlock() if !exists { return fmt.Errorf("extension not found") } + ext.VMMu.Lock() + defer ext.VMMu.Unlock() + if !m.isManagedExtension(extensionID, ext) { + return fmt.Errorf("extension is no longer installed") + } if ext.VM == nil { return nil } - ext.VMMu.Lock() - defer ext.VMMu.Unlock() if err := runCleanupLocked(ext); err != nil { + if IsRuntimeUnsafeError(err) { + quarantineRuntimeLocked(ext, ext.VM, err) + } GoLog("[Extension] Cleanup error for %s: %v\n", extensionID, err) return err } @@ -893,21 +965,27 @@ func (m *extensionManager) UnloadAllExtensions() { } func (m *extensionManager) InvokeAction(extensionID string, actionName string) (map[string]any, error) { - m.mu.Lock() - defer m.mu.Unlock() - + m.mu.RLock() ext, exists := m.extensions[extensionID] + enabled := exists && ext.Enabled + m.mu.RUnlock() if !exists { return nil, fmt.Errorf("extension not found: %s", extensionID) } - if !ext.Enabled { + if !enabled { return nil, fmt.Errorf("extension is disabled") } - vm, err := ext.lockReadyVM() - if err != nil { + ext.VMMu.Lock() + if !m.isManagedExtensionEnabled(extensionID, ext) { + ext.VMMu.Unlock() + return nil, fmt.Errorf("extension is disabled or no longer installed") + } + if err := ensureRuntimeReadyLocked(ext, true); err != nil { + ext.VMMu.Unlock() return nil, err } + vm := ext.VM defer ext.VMMu.Unlock() // Merge extension return values onto the top-level JSON object so Flutter can read diff --git a/go_backend/extension_manager_runtime.go b/go_backend/extension_manager_runtime.go index 798bc6b6..de6243ff 100644 --- a/go_backend/extension_manager_runtime.go +++ b/go_backend/extension_manager_runtime.go @@ -10,6 +10,50 @@ import ( "github.com/dop251/goja" ) +// extensionLifecycleTimeout bounds extension code that runs outside a normal +// provider/action request. A lifecycle callback is still arbitrary extension +// JavaScript, so it needs the same interrupt/quarantine contract as a regular +// invocation. Tests may shorten this value; production keeps the normal JS +// request budget. +var extensionLifecycleTimeout = DefaultJSTimeout + +func runExtensionLifecycleCall( + vm *goja.Runtime, + call func() (goja.Value, error), +) (goja.Value, error) { + return runGojaCallWithTimeoutAndRecover(vm, call, extensionLifecycleTimeout) +} + +// discardLifecycleRuntime releases a runtime whose lifecycle execution failed. +// An interrupted Go callback may still be using the VM after the timeout +// helper returns, so unsafe runtimes must remain quarantined until the helper's +// completion signal closes. Safe failures can be closed immediately. +func discardLifecycleRuntime( + ext *loadedExtension, + vm *goja.Runtime, + runtime *extensionRuntime, + err error, +) { + if IsRuntimeUnsafeError(err) { + if ext != nil && ext.VM == vm { + quarantineRuntimeLocked(ext, vm, err) + } else { + registerQuarantinedRuntime(ext, runtime, runtimeCompletion(err)) + } + return + } + + if runtime != nil { + runtime.closeStorageFlusher() + } + if ext != nil && ext.VM == vm { + ext.VM = nil + ext.runtime = nil + ext.indexProgram = nil + ext.initialized = false + } +} + func initializeVMLocked(ext *loadedExtension) error { ext.VM = nil ext.runtime = nil @@ -21,10 +65,12 @@ func initializeVMLocked(ext *loadedExtension) error { indexPath := filepath.Join(ext.SourceDir, "index.js") jsCode, err := os.ReadFile(indexPath) if err != nil { + ext.VM = nil return fmt.Errorf("failed to read index.js: %w", err) } indexProgram, err := goja.Compile(indexPath, string(jsCode), false) if err != nil { + ext.VM = nil return fmt.Errorf("failed to compile extension code: %w", err) } ext.indexProgram = indexProgram @@ -54,12 +100,16 @@ func initializeVMLocked(ext *loadedExtension) error { return goja.Undefined() }) - _, err = vm.RunProgram(indexProgram) + _, err = runExtensionLifecycleCall(vm, func() (goja.Value, error) { + return vm.RunProgram(indexProgram) + }) if err != nil { + discardLifecycleRuntime(ext, vm, runtime, err) return fmt.Errorf("failed to execute extension code: %w", err) } if registeredExtension == nil || goja.IsUndefined(registeredExtension) { + discardLifecycleRuntime(ext, vm, runtime, fmt.Errorf("extension did not call registerExtension()")) return fmt.Errorf("extension did not call registerExtension()") } @@ -121,20 +171,22 @@ func newIsolatedExtensionRuntime(ext *loadedExtension) (*goja.Runtime, *extensio return goja.Undefined() }) - if _, err := vm.RunProgram(indexProgram); err != nil { - runtime.closeStorageFlusher() + if _, err := runExtensionLifecycleCall(vm, func() (goja.Value, error) { + return vm.RunProgram(indexProgram) + }); err != nil { + discardLifecycleRuntime(ext, vm, runtime, err) return nil, nil, fmt.Errorf("failed to execute extension code: %w", err) } if registeredExtension == nil || goja.IsUndefined(registeredExtension) { - runtime.closeStorageFlusher() + discardLifecycleRuntime(ext, vm, runtime, fmt.Errorf("extension did not call registerExtension()")) return nil, nil, fmt.Errorf("extension did not call registerExtension()") } settings := getExtensionInitSettings(ext.ID) if len(settings) > 0 { if err := initializeExtensionRuntimeWithSettings(vm, ext.ID, settings); err != nil { - runtime.closeStorageFlusher() + discardLifecycleRuntime(ext, vm, runtime, err) return nil, nil, err } } @@ -203,7 +255,12 @@ func releaseIsolatedExtensionRuntime( } if cleanupSafe { - if cleanupErr := runCleanupOnVM(vm); cleanupErr != nil { + cleanupErr := runCleanupOnVM(vm) + if IsRuntimeUnsafeError(cleanupErr) { + registerQuarantinedRuntime(ext, runtime, runtimeCompletion(cleanupErr)) + return + } + if cleanupErr != nil { GoLog("[Extension:%s] isolated download cleanup failed: %v\n", ext.ID, cleanupErr) } } @@ -274,7 +331,12 @@ func drainIsolatedRuntimePool(ext *loadedExtension) { ext.isolatedPoolMu.Unlock() for _, handle := range pool { - if cleanupErr := runCleanupOnVM(handle.vm); cleanupErr != nil { + cleanupErr := runCleanupOnVM(handle.vm) + if IsRuntimeUnsafeError(cleanupErr) { + registerQuarantinedRuntime(ext, handle.runtime, runtimeCompletion(cleanupErr)) + continue + } + if cleanupErr != nil { GoLog("[Extension:%s] isolated pool cleanup failed: %v\n", ext.ID, cleanupErr) } if handle.runtime != nil { @@ -333,7 +395,9 @@ func initializeExtensionRuntimeWithSettings( })() `, string(settingsJSON)) - result, err := vm.RunString(script) + result, err := runExtensionLifecycleCall(vm, func() (goja.Value, error) { + return vm.RunString(script) + }) if err != nil { GoLog("[Extension] Initialize error for %s: %v\n", extensionID, err) return err @@ -367,6 +431,9 @@ func initializeExtensionWithSettingsLocked( if err := initializeExtensionRuntimeWithSettings(ext.VM, ext.ID, settings); err != nil { ext.Error = err.Error() ext.Enabled = false + if IsRuntimeUnsafeError(err) { + quarantineRuntimeLocked(ext, ext.VM, err) + } return err } @@ -406,7 +473,9 @@ func runCleanupOnVM(vm *goja.Runtime) error { })() ` - result, err := vm.RunString(script) + result, err := runExtensionLifecycleCall(vm, func() (goja.Value, error) { + return vm.RunString(script) + }) if err != nil { return err } @@ -429,8 +498,20 @@ func runCleanupOnVM(vm *goja.Runtime) error { func teardownVMLocked(ext *loadedExtension) { drainIsolatedRuntimePool(ext) + // Preserve writes made before cleanup even when the cleanup callback becomes + // unresponsive and its VM has to remain quarantined. + if ext.runtime != nil { + if err := ext.runtime.flushStorageNow(); err != nil { + GoLog("[Extension] Failed to flush storage before cleanup for %s: %v\n", ext.ID, err) + } + } + vm := ext.VM if err := runCleanupLocked(ext); err != nil { GoLog("[Extension] Error calling cleanup for %s: %v\n", ext.ID, err) + if IsRuntimeUnsafeError(err) { + quarantineRuntimeLocked(ext, vm, err) + return + } } if ext.runtime != nil { if err := ext.runtime.flushStorageNow(); err != nil { diff --git a/go_backend/extension_signed_session.go b/go_backend/extension_signed_session.go index b980794c..360744ac 100644 --- a/go_backend/extension_signed_session.go +++ b/go_backend/extension_signed_session.go @@ -2,6 +2,7 @@ package gobackend import ( "bytes" + "context" "crypto/hmac" "crypto/rand" "crypto/sha256" @@ -38,11 +39,17 @@ var ( pendingSignedSessionGrants = make(map[string]string) pendingSignedSessionGrantsMu sync.Mutex signedSessionCoordinators sync.Map - signedSessionRetryWait = time.Sleep - signedSessionProviderWait = sleepRetry - signedSessionRequestNow = time.Now + // signedSessionRetryWait is retained as a test hook for callers that used + // the old duration-only seam. Production waits use the context-aware hook + // below; a non-nil legacy hook short-circuits the delay in tests. + signedSessionRetryWait func(time.Duration) + signedSessionRetryWaitContext = sleepRetry + signedSessionProviderWait = sleepRetry + signedSessionRequestNow = time.Now ) +const signedSessionExchangeTimeout = DefaultJSTimeout + var sessionHintPattern = regexp.MustCompile(`^[0-9a-f]{32}$`) type signedSessionHints struct { @@ -109,6 +116,45 @@ type signedSessionCoordinator struct { pendingExtensionIDs map[string]struct{} completedGrantHash string blockedGeneration string + clearGeneration uint64 + + // exchangeInFlight serializes grant exchanges without keeping mu held over + // HTTP or Retry-After backoff. Waiters observe the completion channel and + // retry their state check after the owner commits or fails. + exchangeInFlight bool + exchangeDone chan struct{} +} + +func (c *signedSessionCoordinator) beginExchange(ctx context.Context) (func(), error) { + if ctx == nil { + ctx = context.Background() + } + for { + c.mu.Lock() + if !c.exchangeInFlight { + c.exchangeInFlight = true + c.exchangeDone = make(chan struct{}) + done := c.exchangeDone + c.mu.Unlock() + return func() { + c.mu.Lock() + if c.exchangeInFlight && c.exchangeDone == done { + c.exchangeInFlight = false + c.exchangeDone = nil + close(done) + } + c.mu.Unlock() + }, nil + } + done := c.exchangeDone + c.mu.Unlock() + + select { + case <-done: + case <-ctx.Done(): + return nil, ctx.Err() + } + } } func (r *extensionRuntime) signedSessionCoordinator(config SignedSessionConfig) (*signedSessionCoordinator, error) { @@ -538,6 +584,11 @@ func (r *extensionRuntime) signedSessionClear(call goja.FunctionCall) goja.Value if err := r.saveSignedSession(config, record); err != nil { return r.vm.ToValue(map[string]any{"success": false, "error": err.Error()}) } + // Invalidate exchanges that released the coordinator lock while performing + // HTTP. A response that started before this explicit clear must never + // resurrect the just-cleared credentials. + coordinator.clearGeneration++ + coordinator.completedGrantHash = "" coordinator.clearBlockedGeneration() coordinator.clearChallenge() ClearPendingAuthRequest(r.extensionID) @@ -560,7 +611,9 @@ func (r *extensionRuntime) signedSessionCompleteGrant(call goja.FunctionCall) go if grant == "" { return r.vm.ToValue(map[string]any{"success": false, "error": "no pending grant"}) } - if err := r.exchangeSignedSessionGrant(grant); err != nil { + ctx, cancel := r.signedSessionExchangeContext() + defer cancel() + if err := r.exchangeSignedSessionGrantContext(ctx, grant); err != nil { return r.vm.ToValue(map[string]any{"success": false, "error": err.Error()}) } pendingSignedSessionGrantsMu.Lock() @@ -571,23 +624,87 @@ func (r *extensionRuntime) signedSessionCompleteGrant(call goja.FunctionCall) go } func (r *extensionRuntime) exchangeSignedSessionGrant(grant string) error { + ctx, cancel := r.signedSessionExchangeContext() + defer cancel() + return r.exchangeSignedSessionGrantContext(ctx, grant) +} + +func (r *extensionRuntime) signedSessionExchangeContext() (context.Context, context.CancelFunc) { + parent := context.Background() + if r != nil { + if itemID := r.getActiveDownloadItemID(); itemID != "" { + parent = downloadCancelContext(itemID) + } else if requestID := r.getActiveRequestID(); requestID != "" { + parent = extensionRequestCancelContext(requestID) + } + } + return context.WithTimeout(parent, signedSessionExchangeTimeout) +} + +func waitSignedSessionRetry(ctx context.Context, delay time.Duration) error { + if ctx == nil { + ctx = context.Background() + } + // Keep the old duration-only hook useful for existing package tests without + // allowing production to fall back to an uninterruptible time.Sleep. The + // default hook is nil; tests install an immediate recorder/no-op here. + if legacyWait := signedSessionRetryWait; legacyWait != nil { + done := make(chan struct{}) + go func() { + legacyWait(delay) + close(done) + }() + select { + case <-done: + return nil + case <-ctx.Done(): + return ctx.Err() + } + } + return signedSessionRetryWaitContext(ctx, delay) +} + +func (r *extensionRuntime) exchangeSignedSessionGrantContext(ctx context.Context, grant string) error { + if r == nil || r.manifest == nil || r.manifest.SignedSession == nil { + return fmt.Errorf("signedSession is not configured") + } + if r.httpClient == nil { + return fmt.Errorf("signed-session exchange HTTP client is unavailable") + } config := signedSessionConfigWithDefaults(r.manifest.SignedSession) coordinator, err := r.signedSessionCoordinator(config) if err != nil { return err } + if ctx == nil { + ctx = context.Background() + } coordinator.mu.Lock() - defer coordinator.mu.Unlock() - return r.exchangeSignedSessionGrantLocked(config, coordinator, grant) + clearGeneration := coordinator.clearGeneration + coordinator.mu.Unlock() + release, err := coordinator.beginExchange(ctx) + if err != nil { + return err + } + defer release() + return r.exchangeSignedSessionGrantLocked(ctx, config, coordinator, clearGeneration, grant) } func (r *extensionRuntime) exchangeSignedSessionGrantLocked( + ctx context.Context, config SignedSessionConfig, coordinator *signedSessionCoordinator, + clearGeneration uint64, grant string, ) error { + coordinator.mu.Lock() + if coordinator.clearGeneration != clearGeneration { + coordinator.mu.Unlock() + return fmt.Errorf("signed-session exchange was superseded by session clear") + } record, err := r.loadSignedSession(config) if err != nil { + coordinator.mu.Unlock() return err } grantHashBytes := sha256.Sum256([]byte(grant)) @@ -599,10 +716,12 @@ func (r *extensionRuntime) exchangeSignedSessionGrantLocked( signedSessionRecordIsUsable(record) { coordinator.clearBlockedGeneration() coordinator.clearChallenge() + coordinator.mu.Unlock() return nil } endpoint, err := signedSessionURL(config, config.Endpoints.Exchange) if err != nil { + coordinator.mu.Unlock() return err } payload := map[string]any{ @@ -612,9 +731,11 @@ func (r *extensionRuntime) exchangeSignedSessionGrantLocked( "platform": config.Platform, } body, _ := json.Marshal(payload) + coordinator.mu.Unlock() + var respBody []byte for attempt := 1; attempt <= signedSessionExchangeMaxAttempts; attempt++ { - req, requestErr := http.NewRequest(http.MethodPost, endpoint, bytes.NewReader(body)) + req, requestErr := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) if requestErr != nil { return requestErr } @@ -646,7 +767,9 @@ func (r *extensionRuntime) exchangeSignedSessionGrantLocked( attempt+1, signedSessionExchangeMaxAttempts, ) - signedSessionRetryWait(retryAfter) + if waitErr := waitSignedSessionRetry(ctx, retryAfter); waitErr != nil { + return waitErr + } continue } if resp.StatusCode < 200 || resp.StatusCode >= 300 { @@ -665,6 +788,30 @@ func (r *extensionRuntime) exchangeSignedSessionGrantLocked( if exchanged.SessionID == "" || exchanged.SessionSecret == "" || exchanged.ExpiresAt == "" { return fmt.Errorf("session exchange response missing session fields") } + + coordinator.mu.Lock() + defer coordinator.mu.Unlock() + if coordinator.clearGeneration != clearGeneration { + return fmt.Errorf("signed-session exchange was superseded by session clear") + } + latest, err := r.loadSignedSession(config) + if err != nil { + return err + } + // Another exchange may have completed while this request was in flight. + // Never overwrite that newer shared session with a stale response. + if coordinator.completedGrantHash == grantHash && + signedSessionRecordIsUsable(latest) { + coordinator.clearBlockedGeneration() + coordinator.clearChallenge() + return nil + } + if signedSessionRecordIsUsable(latest) && !sameSignedSession(latest, record) { + coordinator.clearBlockedGeneration() + coordinator.clearChallenge() + return nil + } + record = latest record.SessionID = exchanged.SessionID record.SessionSecret = exchanged.SessionSecret record.ExpiresAt = exchanged.ExpiresAt diff --git a/go_backend/extension_timeout.go b/go_backend/extension_timeout.go index 1c28268b..6c433c3c 100644 --- a/go_backend/extension_timeout.go +++ b/go_backend/extension_timeout.go @@ -2,6 +2,7 @@ package gobackend import ( "context" + "errors" "fmt" "runtime/debug" "sync" @@ -167,19 +168,21 @@ func runGojaCallWithTimeoutContextAndRecover(ctx context.Context, vm *goja.Runti } func IsRuntimeUnsafeError(err error) bool { - jsErr, ok := err.(*JSExecutionError) - return ok && jsErr.RuntimeUnsafe + var jsErr *JSExecutionError + return errors.As(err, &jsErr) && jsErr.RuntimeUnsafe } func runtimeCompletion(err error) <-chan struct{} { - if jsErr, ok := err.(*JSExecutionError); ok && jsErr.RuntimeUnsafe { + var jsErr *JSExecutionError + if errors.As(err, &jsErr) && jsErr.RuntimeUnsafe { return jsErr.runtimeDone } return nil } func IsTimeoutError(err error) bool { - if jsErr, ok := err.(*JSExecutionError); ok { + var jsErr *JSExecutionError + if errors.As(err, &jsErr) { return jsErr.IsTimeout } return false diff --git a/ios/Runner/AppDelegate.swift b/ios/Runner/AppDelegate.swift index e98d830f..bad78c30 100644 --- a/ios/Runner/AppDelegate.swift +++ b/ios/Runner/AppDelegate.swift @@ -28,8 +28,9 @@ import Gobackend /// Pending Flutter result for the native folder picker private var pendingDirectoryPickerResult: FlutterResult? - /// Whether a download queue is active; while true a background task is - /// started on each background entry to extend execution time. Main-thread only. + /// Whether a download queue is active; while true a background assertion + /// is acquired before work starts and renewed on later background entries. + /// Main-thread only. private var downloadsActive = false private var downloadBackgroundTask: UIBackgroundTaskIdentifier = .invalid @@ -318,6 +319,10 @@ import Gobackend switch call.method { case "beginBackgroundDownloadTask": downloadsActive = true + // Request the assertion before the long-running Go call starts. + // Waiting for applicationDidEnterBackground is too late: iOS may + // suspend the process before the assertion is granted. + beginBackgroundDownloadTask() result(nil) return case "endBackgroundDownloadTask": @@ -419,6 +424,16 @@ import Gobackend downloadBackgroundTask = UIApplication.shared.beginBackgroundTask( withName: "SpotiFLACDownloads" ) { [weak self] in + if self?.downloadsActive == true { + // Flutter channel delivery is asynchronous and iOS may suspend + // us immediately after this callback. Cancel live Go requests + // synchronously first; Dart then persists/requeues the items. + let cancelledItemIDs = GobackendCancelAllActiveDownloads() + self?.backendChannel?.invokeMethod( + "iosBackgroundDownloadExpired", + arguments: cancelledItemIDs + ) + } self?.endBackgroundDownloadTask() } } diff --git a/lib/providers/download_queue_provider.dart b/lib/providers/download_queue_provider.dart index a288b94d..d6ff4a63 100644 --- a/lib/providers/download_queue_provider.dart +++ b/lib/providers/download_queue_provider.dart @@ -362,6 +362,8 @@ class DownloadQueueNotifier extends Notifier { int _queueItemSequence = 0; bool _isLoaded = false; bool _foregroundResumeScheduled = false; + bool _iosBackgroundExecutionExpired = false; + StreamSubscription>? _iosBackgroundExpirationSubscription; final Set _ensuredDirs = {}; final Map> _qualityVariantFileLocks = {}; Future? _appFolderStorageFallback; @@ -423,6 +425,16 @@ class DownloadQueueNotifier extends Notifier { @override DownloadQueueState build() { + if (Platform.isIOS) { + _iosBackgroundExpirationSubscription ??= + PlatformBridge.iosBackgroundDownloadExpirationEvents().listen( + _handleIosBackgroundDownloadExpiration, + ); + ref.onDispose(() { + _iosBackgroundExpirationSubscription?.cancel(); + _iosBackgroundExpirationSubscription = null; + }); + } ref.listen(settingsProvider, (previous, next) { updateSettings(next); if (previous?.downloadNetworkMode != next.downloadNetworkMode) { @@ -487,6 +499,12 @@ class DownloadQueueNotifier extends Notifier { /// Restarts a queue that was deliberately left pending because Android did /// not allow a foreground service to be launched while the app was hidden. void resumePendingDownloadsOnForeground() { + if (Platform.isIOS && _iosBackgroundExecutionExpired) { + _iosBackgroundExecutionExpired = false; + if (state.isPaused) { + state = state.copyWith(isPaused: false); + } + } if (_foregroundResumeScheduled || state.isProcessing || state.isPaused || @@ -504,6 +522,61 @@ class DownloadQueueNotifier extends Notifier { }); } + void _handleIosBackgroundDownloadExpiration(List nativeItemIds) { + if (!Platform.isIOS) return; + final cancelledItemIds = nativeItemIds.toSet(); + final requeueItemIds = cancelledItemIds.where((id) { + final item = state.lookup.byItemId[id]; + return item != null && item.status != DownloadStatus.completed; + }).toSet(); + if (!state.isProcessing && requeueItemIds.isEmpty) return; + + final alreadyForeground = + WidgetsBinding.instance.lifecycleState == AppLifecycleState.resumed; + _log.w( + 'iOS background execution time expired; safely deferring active downloads until foreground', + ); + _iosBackgroundExecutionExpired = true; + if (state.isProcessing && !state.isPaused) { + pauseQueue(persistAcrossRestarts: false); + } + + if (requeueItemIds.isNotEmpty) { + final updatedItems = state.items + .map((item) { + if (!requeueItemIds.contains(item.id)) { + return item; + } + return item.copyWith( + status: DownloadStatus.queued, + progress: 0, + speedMBps: 0, + bytesReceived: 0, + bytesTotal: 0, + ); + }) + .toList(growable: false); + if (state.isProcessing) { + _pausePendingItemIds.addAll(requeueItemIds); + } + state = state.copyWith( + items: updatedItems, + isPaused: true, + currentDownload: null, + ); + } else if (!state.isProcessing) { + state = state.copyWith(isPaused: true, currentDownload: null); + } + unawaited(flushQueuePersistence()); + // The native expiration callback cancels Go synchronously before this + // asynchronous event reaches Dart. If foregrounding won that race, still + // requeue the cancelled item, then immediately let the running queue loop + // continue instead of misclassifying it as a skipped download. + if (alreadyForeground) { + resumePendingDownloadsOnForeground(); + } + } + Future _loadQueueFromStorage() async { if (_isLoaded) return; _isLoaded = true; @@ -1829,6 +1902,13 @@ class DownloadQueueNotifier extends Notifier { } else { _log.i('Queue processing finished'); } + // All per-item futures have completed at this point. Remove any iOS + // expiration guards that arrived after an individual worker's finally + // block, otherwise the safely requeued item would be filtered forever on + // the next queue run. + _pausePendingItemIds.removeWhere( + (id) => state.lookup.byItemId[id]?.status == DownloadStatus.queued, + ); state = state.copyWith(isProcessing: false, currentDownload: null); final hasQueuedItems = state.items.any( diff --git a/lib/providers/download_queue_provider_native_worker.dart b/lib/providers/download_queue_provider_native_worker.dart index d43fb76a..04c97c0b 100644 --- a/lib/providers/download_queue_provider_native_worker.dart +++ b/lib/providers/download_queue_provider_native_worker.dart @@ -984,6 +984,7 @@ extension _DownloadQueueNativeWorker on DownloadQueueNotifier { ) async { final rawItems = snapshot['items']; final rawDelta = snapshot['item_delta']; + final rawDeltas = snapshot['item_deltas']; final itemSnapshots = >[]; if (rawItems is List) { for (final rawItem in rawItems) { @@ -995,6 +996,13 @@ extension _DownloadQueueNativeWorker on DownloadQueueNotifier { if (rawDelta is Map) { itemSnapshots.add(Map.from(rawDelta)); } + if (rawDeltas is List) { + for (final rawItem in rawDeltas) { + if (rawItem is Map) { + itemSnapshots.add(Map.from(rawItem)); + } + } + } if (itemSnapshots.isEmpty) { return; } diff --git a/lib/services/app_state_database.dart b/lib/services/app_state_database.dart index 06d115d9..6de52546 100644 --- a/lib/services/app_state_database.dart +++ b/lib/services/app_state_database.dart @@ -8,7 +8,7 @@ import 'package:spotiflac_android/utils/logger.dart'; final _log = AppLogger('AppStateDb'); const _dbFileName = 'app_state.db'; -const _dbVersion = 3; +const _dbVersion = 4; const _queueTable = 'download_queue_items'; const _recentTable = 'recent_access_items'; @@ -87,12 +87,12 @@ class AppStateDatabase { Future _upgradeDb(Database db, int oldVersion, int newVersion) async { _log.i('Upgrading app state database from v$oldVersion to v$newVersion'); - if (oldVersion < 2) { - await _createPlaybackSessionTable(db); - } if (oldVersion < 3) { await _createRecentStateTable(db); } + if (oldVersion < 4) { + await _migratePlaybackSessionToV4(db); + } } static Future _createRecentStateTable(Database db) { @@ -105,18 +105,88 @@ class AppStateDatabase { } static Future _createPlaybackSessionTable(Database db) { - // Keep this idempotent so an interrupted migration or a database restored - // from an intermediate build can resume v1 -> v2 without losing queue - // state merely because the table was already created. return db.execute(''' CREATE TABLE IF NOT EXISTS $_playbackSessionTable ( id INTEGER PRIMARY KEY CHECK (id = 1), - session_json TEXT NOT NULL, + media_json TEXT NOT NULL, + current_index INTEGER NOT NULL DEFAULT 0, + position_ms INTEGER NOT NULL DEFAULT 0, + shuffle INTEGER NOT NULL DEFAULT 0, + repeat_mode TEXT NOT NULL DEFAULT 'none', updated_at TEXT NOT NULL ) '''); } + static Future _migratePlaybackSessionToV4(Database db) async { + final table = await db.rawQuery( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?", + [_playbackSessionTable], + ); + if (table.isEmpty) { + await _createPlaybackSessionTable(db); + return; + } + + final columns = await db.rawQuery( + 'PRAGMA table_info($_playbackSessionTable)', + ); + if (columns.any((column) => column['name'] == 'media_json')) return; + + Map? legacySession; + final rows = await db.query(_playbackSessionTable, limit: 1); + final raw = rows.isEmpty ? null : rows.first['session_json'] as String?; + if (raw != null && raw.isNotEmpty) { + try { + final decoded = jsonDecode(raw); + if (decoded is Map) { + legacySession = Map.from(decoded); + } + } catch (e) { + _log.w('Discarding unreadable legacy playback session: $e'); + } + } + + const migratedTable = '${_playbackSessionTable}_v4'; + await db.execute('DROP TABLE IF EXISTS $migratedTable'); + await db.execute(''' + CREATE TABLE $migratedTable ( + id INTEGER PRIMARY KEY CHECK (id = 1), + media_json TEXT NOT NULL, + current_index INTEGER NOT NULL DEFAULT 0, + position_ms INTEGER NOT NULL DEFAULT 0, + shuffle INTEGER NOT NULL DEFAULT 0, + repeat_mode TEXT NOT NULL DEFAULT 'none', + updated_at TEXT NOT NULL + ) + '''); + if (legacySession != null) { + await db.insert(migratedTable, _playbackSessionRow(legacySession)); + } + await db.execute('DROP TABLE $_playbackSessionTable'); + await db.execute( + 'ALTER TABLE $migratedTable RENAME TO $_playbackSessionTable', + ); + } + + static Map _playbackSessionRow( + Map session, + ) { + final media = session['media']; + final currentIndex = session['index']; + final positionMs = session['positionMs']; + final repeatMode = session['repeat']; + return { + 'id': 1, + 'media_json': jsonEncode(media is List ? media : const []), + 'current_index': currentIndex is num ? currentIndex.toInt() : 0, + 'position_ms': positionMs is num ? positionMs.toInt() : 0, + 'shuffle': session['shuffle'] == true ? 1 : 0, + 'repeat_mode': repeatMode is String ? repeatMode : 'none', + 'updated_at': DateTime.now().toIso8601String(), + }; + } + Future migrateQueueFromSharedPreferences() async { final prefs = await _prefs; if (prefs.getBool(_queueMigrationKey) == true) { @@ -288,11 +358,20 @@ class AppStateDatabase { final db = await database; final rows = await db.query(_playbackSessionTable, limit: 1); if (rows.isEmpty) return null; - final raw = rows.first['session_json'] as String?; + final row = rows.first; + final raw = row['media_json'] as String?; if (raw == null || raw.isEmpty) return null; try { final decoded = jsonDecode(raw); - if (decoded is Map) return Map.from(decoded); + if (decoded is! List) return null; + return { + 'version': 2, + 'media': decoded, + 'index': (row['current_index'] as num?)?.toInt() ?? 0, + 'positionMs': (row['position_ms'] as num?)?.toInt() ?? 0, + 'shuffle': (row['shuffle'] as num?)?.toInt() == 1, + 'repeat': row['repeat_mode'] as String? ?? 'none', + }; } catch (e) { _log.w('Discarding unreadable playback session: $e'); } @@ -301,11 +380,28 @@ class AppStateDatabase { Future savePlaybackSession(Map session) async { final db = await database; - await db.insert(_playbackSessionTable, { - 'id': 1, - 'session_json': jsonEncode(session), + await db.insert( + _playbackSessionTable, + _playbackSessionRow(session), + conflictAlgorithm: ConflictAlgorithm.replace, + ); + } + + Future updatePlaybackSessionState({ + required int index, + required int positionMs, + required bool shuffle, + required String repeatMode, + }) async { + final db = await database; + final changed = await db.update(_playbackSessionTable, { + 'current_index': index, + 'position_ms': positionMs, + 'shuffle': shuffle ? 1 : 0, + 'repeat_mode': repeatMode, 'updated_at': DateTime.now().toIso8601String(), - }, conflictAlgorithm: ConflictAlgorithm.replace); + }, where: 'id = 1'); + return changed > 0; } Future clearPlaybackSession() async { diff --git a/lib/services/history_database.dart b/lib/services/history_database.dart index b93112b3..31e3847c 100644 --- a/lib/services/history_database.dart +++ b/lib/services/history_database.dart @@ -65,24 +65,39 @@ class HistoryBatchLookupRequest { } class HistoryDatabase { + // The FTS table is a derived, optional index and is initialized lazily after + // the existing schema migration. Keep this contract at v13 because the + // background native writer shares history.db and must accept the same + // user_version without depending on FTS5. static const int schemaVersion = 13; + static const String searchFtsTable = 'history_search_fts'; static final HistoryDatabase instance = HistoryDatabase._init(); static final sqlite.SingleFlightInitializer _database = sqlite.SingleFlightInitializer(); + bool _searchFtsAvailable = false; HistoryDatabase._init(); Future get database { - return _database.getOrCreate( - () => sqlite.openAppDatabase( + return _database.getOrCreate(() async { + final db = await sqlite.openAppDatabase( 'history.db', version: schemaVersion, onCreate: _createDB, onUpgrade: _upgradeDB, - ), - ); + ); + // onCreate normally initializes this derived index. Retry once after + // opening an existing database in case an earlier setup was + // interrupted; unsupported SQLite builds remain on the LIKE fallback. + if (!_searchFtsAvailable) { + _searchFtsAvailable = await _createSearchFts(db); + } + return db; + }); } + bool get searchFtsAvailable => _searchFtsAvailable; + Future _createDB(Database db, int version) async { _log.i('Creating database schema v$version'); @@ -151,6 +166,7 @@ class HistoryDatabase { await _createNormalizedIndexes(db); await _createQueueIndexes(db); await _createPathKeyTable(db); + _searchFtsAvailable = await _createSearchFts(db); _log.i('Database schema created with indexes'); } @@ -258,6 +274,15 @@ class HistoryDatabase { } } + Future _createSearchFts(DatabaseExecutor db) { + return sqlite.createTrigramFtsIndex( + db, + ftsTable: searchFtsTable, + contentTable: 'history', + triggerPrefix: 'history_search_fts', + ); + } + static String normalizeLookupText(String? value) => sqlite.normalizeLookupText(value); diff --git a/lib/services/library_database.dart b/lib/services/library_database.dart index 3c91adae..479272bc 100644 --- a/lib/services/library_database.dart +++ b/lib/services/library_database.dart @@ -16,27 +16,40 @@ final _log = AppLogger('LibraryDatabase'); class LibraryDatabase { static final LibraryDatabase instance = LibraryDatabase._init(); + // The FTS table is a derived, optional index and is initialized lazily after + // the existing schema migration, so it does not require a user_version bump. static const int schemaVersion = 13; static const String legacySourceId = LocalLibraryItem.legacySourceId; static const String visibleLibraryView = 'library_visible'; + static const String searchFtsTable = 'library_search_fts'; static const int audioMetadataScanVersion = 3; static final sqlite.SingleFlightInitializer _database = sqlite.SingleFlightInitializer(); bool _historyAttached = false; + bool _searchFtsAvailable = false; LibraryDatabase._init(); Future get database { - return _database.getOrCreate( - () => sqlite.openAppDatabase( + return _database.getOrCreate(() async { + final db = await sqlite.openAppDatabase( 'local_library.db', version: schemaVersion, onCreate: _createDB, onUpgrade: _upgradeDB, - ), - ); + ); + // onCreate normally initializes this derived index. Retry once after + // opening an existing database in case an earlier setup was + // interrupted; unsupported SQLite builds remain on the LIKE fallback. + if (!_searchFtsAvailable) { + _searchFtsAvailable = await _createSearchFts(db); + } + return db; + }); } + bool get searchFtsAvailable => _searchFtsAvailable; + Future _ensureHistoryAttached(Database db) async { if (_historyAttached) return; await HistoryDatabase.instance.database; @@ -114,6 +127,7 @@ class LibraryDatabase { await _createQueueIndexes(db); await _createPathKeyTable(db); await _createLibrarySources(db); + _searchFtsAvailable = await _createSearchFts(db); _log.i('Library database schema created with indexes'); } @@ -223,6 +237,15 @@ class LibraryDatabase { } } + Future _createSearchFts(DatabaseExecutor db) { + return sqlite.createTrigramFtsIndex( + db, + ftsTable: searchFtsTable, + contentTable: 'library', + triggerPrefix: 'library_search_fts', + ); + } + Future _createLibrarySources(DatabaseExecutor db) async { await db.execute(''' CREATE TABLE IF NOT EXISTS library_sources ( diff --git a/lib/services/library_database_queue_sql.dart b/lib/services/library_database_queue_sql.dart index eaacc70b..78753976 100644 --- a/lib/services/library_database_queue_sql.dart +++ b/lib/services/library_database_queue_sql.dart @@ -365,9 +365,21 @@ extension _LibraryDbQueueSql on LibraryDatabase { ) { final query = LibraryDatabase.normalizeLookupText(request.searchQuery); if (query.isNotEmpty) { - final like = '%${_escapeLikePattern(query)}%'; - where.add("h.search_text LIKE ? ESCAPE '\\'"); - args.add(like); + final ftsQuery = sqlite.ftsPhraseSearchQuery(query); + if (HistoryDatabase.instance.searchFtsAvailable && ftsQuery != null) { + where.add(''' + h.rowid IN ( + SELECT rowid + FROM history_db.history_search_fts + WHERE history_search_fts MATCH ? + ) + '''); + args.add(ftsQuery); + } else { + final like = '%${_escapeLikePattern(query)}%'; + where.add("h.search_text LIKE ? ESCAPE '\\'"); + args.add(like); + } } _appendQueueCommonFilters( where, @@ -396,9 +408,21 @@ extension _LibraryDbQueueSql on LibraryDatabase { ) { final query = LibraryDatabase.normalizeLookupText(request.searchQuery); if (query.isNotEmpty) { - final like = '%${_escapeLikePattern(query)}%'; - where.add("l.search_text LIKE ? ESCAPE '\\'"); - args.add(like); + final ftsQuery = sqlite.ftsPhraseSearchQuery(query); + if (searchFtsAvailable && ftsQuery != null) { + where.add(''' + l.rowid IN ( + SELECT rowid + FROM library_search_fts + WHERE library_search_fts MATCH ? + ) + '''); + args.add(ftsQuery); + } else { + final like = '%${_escapeLikePattern(query)}%'; + where.add("l.search_text LIKE ? ESCAPE '\\'"); + args.add(like); + } } _appendQueueCommonFilters( where, diff --git a/lib/services/music_player_service.dart b/lib/services/music_player_service.dart index becd8aa0..2d4682dc 100644 --- a/lib/services/music_player_service.dart +++ b/lib/services/music_player_service.dart @@ -296,6 +296,9 @@ class MusicPlayerHandler extends BaseAudioHandler DateTime? _lastPositionBroadcastAt; DateTime? _lastPeriodicPersistAt; Future _sessionWriteTail = Future.value(); + int _sessionQueueRevision = 0; + int _scheduledSessionQueueRevision = -1; + int _persistedSessionQueueRevision = -1; static const Duration _positionBroadcastInterval = Duration( milliseconds: 500, ); @@ -653,27 +656,77 @@ class MusicPlayerHandler extends BaseAudioHandler return queued; } - /// Persists queue, current index, and position so a killed process can - /// restore the session paused on next launch. Position updates are throttled - /// by [_handlePositionChanged], while lifecycle/pause writes flush exactly. + void _markSessionQueueChanged() { + _sessionQueueRevision++; + } + + /// Persists the queue only when it changed. Periodic position updates write + /// fixed-size scalar columns, avoiding full queue JSON serialization every + /// ten seconds for large playback sessions. Future _persistSession({Duration? position}) { if (_restoringSession) return Future.value(); if (_media.isEmpty || _index < 0 || _index >= _media.length) { + _scheduledSessionQueueRevision = -1; + _persistedSessionQueueRevision = -1; return _enqueueSessionWrite( AppStateDatabase.instance.clearPlaybackSession, ); } - final session = { - 'version': 1, - 'media': _media.map((m) => m.toJson()).toList(growable: false), - 'index': _index, - 'positionMs': (position ?? Duration.zero).inMilliseconds, - 'shuffle': _shuffle, - 'repeat': _repeatMode.name, - }; - return _enqueueSessionWrite( - () => AppStateDatabase.instance.savePlaybackSession(session), - ); + final queueRevision = _sessionQueueRevision; + final index = _index; + final positionMs = (position ?? Duration.zero).inMilliseconds; + final shuffle = _shuffle; + final repeatMode = _repeatMode.name; + + if (_scheduledSessionQueueRevision != queueRevision) { + final media = _media.map((item) => item.toJson()).toList(growable: false); + _scheduledSessionQueueRevision = queueRevision; + return _enqueueSessionWrite(() async { + try { + await AppStateDatabase.instance.savePlaybackSession({ + 'version': 2, + 'media': media, + 'index': index, + 'positionMs': positionMs, + 'shuffle': shuffle, + 'repeat': repeatMode, + }); + _persistedSessionQueueRevision = queueRevision; + } catch (_) { + if (_scheduledSessionQueueRevision == queueRevision) { + _scheduledSessionQueueRevision = _persistedSessionQueueRevision; + } + rethrow; + } + }); + } + + return _enqueueSessionWrite(() async { + final updated = await AppStateDatabase.instance + .updatePlaybackSessionState( + index: index, + positionMs: positionMs, + shuffle: shuffle, + repeatMode: repeatMode, + ); + if (updated) return; + // A newer queue snapshot is already (or is about to be) scheduled. Do + // not recreate a missing row from this older scalar snapshot with a + // mismatched index; the newer full write will restore it consistently. + if (_sessionQueueRevision != queueRevision) return; + + // Defensive recovery for an externally cleared/corrupted row. This is + // intentionally the only state-only path that serializes the queue. + await AppStateDatabase.instance.savePlaybackSession({ + 'version': 2, + 'media': _media.map((item) => item.toJson()).toList(growable: false), + 'index': index, + 'positionMs': positionMs, + 'shuffle': shuffle, + 'repeat': repeatMode, + }); + _persistedSessionQueueRevision = queueRevision; + }); } Future _currentPositionForPersist() async { @@ -706,6 +759,7 @@ class MusicPlayerHandler extends BaseAudioHandler required int index, required Duration position, required bool shuffle, + bool queueNeedsRewrite = false, AudioServiceRepeatMode repeatMode = AudioServiceRepeatMode.none, }) async { if (items.isEmpty) return; @@ -725,6 +779,14 @@ class MusicPlayerHandler extends BaseAudioHandler _pendingRestorePosition = position > Duration.zero ? position : null; _sourceReady = false; _lastPeriodicPersistAt = null; + _sessionQueueRevision++; + if (queueNeedsRewrite) { + _scheduledSessionQueueRevision = -1; + _persistedSessionQueueRevision = -1; + } else { + _scheduledSessionQueueRevision = _sessionQueueRevision; + _persistedSessionQueueRevision = _sessionQueueRevision; + } queue.add(List.unmodifiable(_queueItems)); mediaItem.add(_media[_index].toMediaItem()); if (position > Duration.zero) { @@ -736,6 +798,9 @@ class MusicPlayerHandler extends BaseAudioHandler } finally { _restoringSession = false; } + if (queueNeedsRewrite) { + await _persistSession(position: position); + } } bool _isCurrentPlayRequest(int generation, PlayableMedia media) { @@ -758,6 +823,7 @@ class MusicPlayerHandler extends BaseAudioHandler _queueItems ..clear() ..addAll(items.map((m) => m.toMediaItem())); + _markSessionQueueChanged(); _recent.clear(); _playHistory.clear(); queue.add(List.unmodifiable(_queueItems)); @@ -774,6 +840,7 @@ class MusicPlayerHandler extends BaseAudioHandler : _media.length; _media.insert(insertAt, item); _queueItems.insert(insertAt, item.toMediaItem()); + _markSessionQueueChanged(); for (var i = 0; i < _recent.length; i++) { if (_recent[i] >= insertAt) _recent[i]++; @@ -808,6 +875,7 @@ class MusicPlayerHandler extends BaseAudioHandler } at++; } + _markSessionQueueChanged(); queue.add(List.unmodifiable(_queueItems)); _broadcastState(); unawaited(_persistSession(position: playbackState.value.position)); @@ -825,6 +893,7 @@ class MusicPlayerHandler extends BaseAudioHandler final qi = _queueItems.removeAt(oldIndex); _media.insert(newIndex, media); _queueItems.insert(newIndex, qi); + _markSessionQueueChanged(); if (_index == oldIndex) { _index = newIndex; @@ -1127,6 +1196,8 @@ class MusicPlayerHandler extends BaseAudioHandler _recent.clear(); _playHistory.clear(); _pendingRestorePosition = null; + _scheduledSessionQueueRevision = -1; + _persistedSessionQueueRevision = -1; // An explicit stop ends the session for good; nothing to restore later. await _enqueueSessionWrite(AppStateDatabase.instance.clearPlaybackSession); // A stopped session has no current item; this also hides the mini player. @@ -1236,6 +1307,7 @@ class MusicPlayerHandler extends BaseAudioHandler _queueItems ..clear() ..addAll(kept.map((m) => m.toMediaItem())); + _markSessionQueueChanged(); _recent.clear(); _playHistory.clear(); queue.add(List.unmodifiable(_queueItems)); @@ -1376,6 +1448,7 @@ Future restorePersistedPlaybackSession() async { index: index, position: position, shuffle: session['shuffle'] == true, + queueNeedsRewrite: items.length != rawMedia.length, repeatMode: AudioServiceRepeatMode.values.firstWhere( (mode) => mode.name == session['repeat'], orElse: () => AudioServiceRepeatMode.none, diff --git a/lib/services/platform_bridge.dart b/lib/services/platform_bridge.dart index 5a7887f3..b450e11d 100644 --- a/lib/services/platform_bridge.dart +++ b/lib/services/platform_bridge.dart @@ -181,6 +181,9 @@ class PlatformBridge { StreamController.broadcast(); static final StreamController _libraryStorageEvents = StreamController.broadcast(); + static final StreamController> + _iosBackgroundDownloadExpirationEvents = + StreamController>.broadcast(); static bool _backendEventHandlerInstalled = false; static bool get supportsCoreBackend => Platform.isAndroid || Platform.isIOS; @@ -198,6 +201,11 @@ class PlatformBridge { return _libraryStorageEvents.stream; } + static Stream> iosBackgroundDownloadExpirationEvents() { + _ensureBackendEventHandler(); + return _iosBackgroundDownloadExpirationEvents.stream; + } + static void _ensureBackendEventHandler() { if (_backendEventHandlerInstalled) return; _backendEventHandlerInstalled = true; @@ -220,6 +228,24 @@ class PlatformBridge { case 'libraryStorageChanged': _libraryStorageEvents.add(null); return null; + case 'iosBackgroundDownloadExpired': + final raw = call.arguments; + var itemIds = const []; + try { + final decoded = raw is String ? jsonDecode(raw) : raw; + if (decoded is List) { + itemIds = decoded + .map((value) => value.toString().trim()) + .where((value) => value.isNotEmpty) + .toSet() + .toList(growable: false); + } + } catch (_) { + // Older/native-mismatched builds may send no payload. The queue + // still pauses its currently visible active items below. + } + _iosBackgroundDownloadExpirationEvents.add(itemIds); + return null; default: return null; } diff --git a/lib/services/sqlite_helpers.dart b/lib/services/sqlite_helpers.dart index 39d6d7b8..b8366f7e 100644 --- a/lib/services/sqlite_helpers.dart +++ b/lib/services/sqlite_helpers.dart @@ -65,6 +65,11 @@ Future openAppDatabase( if (foreignKeys) { await db.execute('PRAGMA foreign_keys = ON'); } + // History/library use INSERT OR REPLACE extensively. SQLite only fires + // delete triggers for REPLACE when recursive_triggers is enabled; the + // FTS external-content delete trigger needs that event to remove the old + // rowid instead of accumulating unreachable index entries. + await db.execute('PRAGMA recursive_triggers = ON'); if (incrementalAutoVacuum) { final tables = await db.rawQuery(''' SELECT 1 @@ -93,6 +98,101 @@ String normalizeLookupText(String? value) { return (value ?? '').trim().toLowerCase(); } +/// Returns a literal phrase suitable for the trigram FTS5 MATCH operator. +/// +/// The trigram tokenizer cannot answer one- or two-character searches, so +/// callers should use their compatibility fallback when this returns null. +/// Quoting and escaping the value keeps user-entered FTS operators literal. +String? ftsPhraseSearchQuery(String value) { + if (value.runes.length < 3 || value.contains('\u0000')) return null; + return '"${value.replaceAll('"', '""')}"'; +} + +/// Creates an external-content FTS5 index that preserves substring search +/// semantics through SQLite's trigram tokenizer. +/// +/// FTS5 is an optional SQLite extension on some platform/database builds, so +/// callers must retain their existing query fallback when this returns false. +/// The index is external-content: the source table remains authoritative and +/// these triggers keep the index synchronized for every insert/update/delete, +/// including writes that happen outside the Dart repository methods. +Future createTrigramFtsIndex( + DatabaseExecutor db, { + required String ftsTable, + required String contentTable, + required String triggerPrefix, +}) async { + try { + final existingIndex = await db.rawQuery( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ? LIMIT 1", + [ftsTable], + ); + final existingTriggers = await db.rawQuery( + "SELECT name FROM sqlite_master WHERE type = 'trigger' AND name IN (?, ?, ?)", + ['${triggerPrefix}_ai', '${triggerPrefix}_ad', '${triggerPrefix}_au'], + ); + final needsRebuild = existingIndex.isEmpty || existingTriggers.length != 3; + await db.execute(''' + CREATE VIRTUAL TABLE IF NOT EXISTS $ftsTable USING fts5( + search_text, + content='$contentTable', + content_rowid='rowid', + tokenize='trigram' + ) + '''); + await db.execute(''' + CREATE TRIGGER IF NOT EXISTS ${triggerPrefix}_ai + AFTER INSERT ON $contentTable + BEGIN + INSERT INTO $ftsTable(rowid, search_text) + VALUES (new.rowid, COALESCE(new.search_text, '')); + END + '''); + await db.execute(''' + CREATE TRIGGER IF NOT EXISTS ${triggerPrefix}_ad + AFTER DELETE ON $contentTable + BEGIN + INSERT INTO $ftsTable($ftsTable, rowid, search_text) + VALUES ('delete', old.rowid, COALESCE(old.search_text, '')); + END + '''); + await db.execute(''' + CREATE TRIGGER IF NOT EXISTS ${triggerPrefix}_au + AFTER UPDATE OF search_text ON $contentTable + BEGIN + INSERT INTO $ftsTable($ftsTable, rowid, search_text) + VALUES ('delete', old.rowid, COALESCE(old.search_text, '')); + INSERT INTO $ftsTable(rowid, search_text) + VALUES (new.rowid, COALESCE(new.search_text, '')); + END + '''); + + // Rebuild a new or partially-created index. Avoid doing this on every app + // start: the external-content table is already kept current by triggers. + if (needsRebuild) { + await db.rawInsert("INSERT INTO $ftsTable($ftsTable) VALUES (?)", [ + 'rebuild', + ]); + } + return true; + } catch (error) { + _log.w( + 'FTS5 index unavailable for $contentTable; using LIKE fallback: $error', + ); + // Do not leave a half-created index/triggers behind. This makes a later + // retry deterministic and never compromises the authoritative table. + try { + await db.execute('DROP TRIGGER IF EXISTS ${triggerPrefix}_ai'); + await db.execute('DROP TRIGGER IF EXISTS ${triggerPrefix}_ad'); + await db.execute('DROP TRIGGER IF EXISTS ${triggerPrefix}_au'); + await db.execute('DROP TABLE IF EXISTS $ftsTable'); + } catch (cleanupError) { + _log.w('Failed to clean up partial FTS5 index $ftsTable: $cleanupError'); + } + return false; + } +} + Future addColumnIfMissing( Database db, String table, diff --git a/test/models_and_utils_test.dart b/test/models_and_utils_test.dart index 080efa88..aeaa48c6 100644 --- a/test/models_and_utils_test.dart +++ b/test/models_and_utils_test.dart @@ -169,18 +169,76 @@ void main() { final source = File( 'lib/services/app_state_database.dart', ).readAsStringSync(); + final playerSource = File( + 'lib/services/music_player_service.dart', + ).readAsStringSync(); - test('v1 to v2 tolerates an existing playback session table', () { + test('v4 normalizes playback queue and scalar state', () { + expect(source, contains('const _dbVersion = 4;')); + expect(source, contains('await _migratePlaybackSessionToV4(db);')); + expect(source, contains('media_json TEXT NOT NULL')); + expect(source, contains('position_ms INTEGER NOT NULL DEFAULT 0')); + expect(source, contains('updatePlaybackSessionState')); + }); + + test('periodic playback updates do not serialize an unchanged queue', () { + final scalarUpdate = RegExp( + r'updatePlaybackSessionState\([\s\S]*?if \(updated\) return;', + ).firstMatch(playerSource); + + expect(scalarUpdate, isNotNull); + expect(scalarUpdate!.group(0), isNot(contains("'media':"))); + expect(playerSource, contains('_scheduledSessionQueueRevision')); + }); + }); + + group('iOS background download lifecycle', () { + final appDelegateSource = File( + 'ios/Runner/AppDelegate.swift', + ).readAsStringSync(); + final queueProviderSource = File( + 'lib/providers/download_queue_provider.dart', + ).readAsStringSync(); + + test('acquires background time before starting long-running work', () { + final beginCase = RegExp( + r'case "beginBackgroundDownloadTask":[\s\S]*?result\(nil\)', + ).firstMatch(appDelegateSource); + expect(beginCase, isNotNull); expect( - source, - contains('CREATE TABLE IF NOT EXISTS \$_playbackSessionTable'), + beginCase!.group(0)!.indexOf('beginBackgroundDownloadTask()'), + lessThan(beginCase.group(0)!.indexOf('result(nil)')), + ); + }); + + test('expiration safely defers and foreground resumes the queue', () { + expect(appDelegateSource, contains('iosBackgroundDownloadExpired')); + final expirationHandler = RegExp( + r'if self\?\.downloadsActive == true \{[\s\S]*?\n\s*\}', + ).firstMatch(appDelegateSource); + expect(expirationHandler, isNotNull); + expect( + expirationHandler! + .group(0)! + .indexOf('GobackendCancelAllActiveDownloads()'), + lessThan( + expirationHandler.group(0)!.indexOf('iosBackgroundDownloadExpired'), + ), ); expect( - RegExp( - r'if \(oldVersion < 2\)\s*\{\s*' - r'await _createPlaybackSessionTable\(db\);', - ).hasMatch(source), - isTrue, + queueProviderSource, + contains('pauseQueue(persistAcrossRestarts: false)'), + ); + expect(queueProviderSource, contains('_iosBackgroundExecutionExpired')); + expect(queueProviderSource, contains('final requeueItemIds =')); + expect( + queueProviderSource, + contains('if (!state.isProcessing && requeueItemIds.isEmpty) return;'), + ); + expect(queueProviderSource, contains('if (alreadyForeground) {')); + expect( + queueProviderSource, + contains('resumePendingDownloadsOnForeground();'), ); }); }); @@ -193,6 +251,13 @@ void main() { final historyDatabaseSource = File( 'lib/services/history_database.dart', ).readAsStringSync(); + final workerSnapshotSource = File( + 'android/app/src/main/kotlin/com/zarz/spotiflac/' + 'DownloadServiceSnapshot.kt', + ).readAsStringSync(); + final nativeWorkerProviderSource = File( + 'lib/providers/download_queue_provider_native_worker.dart', + ).readAsStringSync(); int kotlinConstant(String name) { final match = RegExp( @@ -216,6 +281,27 @@ void main() { ); }); + test('polls one progress delta stream and preserves concurrent items', () { + expect( + RegExp( + r'Gobackend\.getAllDownloadProgressDelta\(', + ).allMatches(workerSnapshotSource), + hasLength(1), + ); + expect( + workerSnapshotSource, + isNot(contains('Gobackend.getAllDownloadProgress()')), + ); + expect(workerSnapshotSource, contains('"item_deltas"')); + expect( + workerSnapshotSource, + contains('progressItemIds = orderedItemIds'), + ); + expect(workerSnapshotSource, contains('progressCoordinatorEpoch')); + expect(workerSnapshotSource, contains('nativeWorkerProgressEpoch.get()')); + expect(nativeWorkerProviderSource, contains("snapshot['item_deltas']")); + }); + Set historyTableColumns(String source) { final match = RegExp( r'CREATE TABLE(?: IF NOT EXISTS)? history\s*\(([\s\S]*?)\n\s*\)', diff --git a/test/sqlite_helpers_test.dart b/test/sqlite_helpers_test.dart index f13d58dc..a0f48422 100644 --- a/test/sqlite_helpers_test.dart +++ b/test/sqlite_helpers_test.dart @@ -87,5 +87,21 @@ void main() { ); expect(autoVacuumMatches.single.start, lessThan(onCreateIndex)); }); + + test('enables delete triggers for replace-backed FTS synchronization', () { + expect(source, contains('PRAGMA recursive_triggers = ON')); + }); + }); + + group('FTS5 search query', () { + test( + 'quotes literal input and leaves unsupported short terms to fallback', + () { + expect(ftsPhraseSearchQuery('ab'), isNull); + expect(ftsPhraseSearchQuery('beat'), '"beat"'); + expect(ftsPhraseSearchQuery('foo"bar'), '"foo""bar"'); + expect(ftsPhraseSearchQuery('a\u0000b'), isNull); + }, + ); }); }