diff --git a/android/app/src/main/kotlin/com/zarz/spotiflac/NativeFinalizerFFmpeg.kt b/android/app/src/main/kotlin/com/zarz/spotiflac/NativeFinalizerFFmpeg.kt index e586a56b..6da15cce 100644 --- a/android/app/src/main/kotlin/com/zarz/spotiflac/NativeFinalizerFFmpeg.kt +++ b/android/app/src/main/kotlin/com/zarz/spotiflac/NativeFinalizerFFmpeg.kt @@ -153,7 +153,7 @@ internal fun NativeDownloadFinalizer.withFFmpegCommandPump( val pump = Thread { while (running.get()) { try { - val raw = Gobackend.getAllPendingFFmpegCommandsJSON() + val raw = Gobackend.waitForPendingFFmpegCommandsJSON(1_000L) val commands = org.json.JSONArray(raw) for (index in 0 until commands.length()) { val command = commands.optJSONObject(index) ?: continue @@ -190,13 +190,6 @@ internal fun NativeDownloadFinalizer.withFFmpegCommandPump( } } catch (_: Exception) { } - try { - Thread.sleep(100) - } catch (_: InterruptedException) { - // Keep pumping until `running` flips: on cancel the Go call - // may still be waiting for a result for an in-flight - // command, and it is delivered as failed above. - } } } pump.isDaemon = true diff --git a/go_backend/exports_extensions.go b/go_backend/exports_extensions.go index b6e0daec..90a31ca7 100644 --- a/go_backend/exports_extensions.go +++ b/go_backend/exports_extensions.go @@ -717,7 +717,7 @@ func GetAllPendingFFmpegCommandsJSON() (string, error) { commands := make([]map[string]any, 0) for cmdID, cmd := range ffmpegCommands { - if !cmd.Completed { + if !cmd.Completed && !cmd.Claimed { commands = append(commands, map[string]any{ "command_id": cmdID, "extension_id": cmd.ExtensionID, @@ -729,6 +729,47 @@ func GetAllPendingFFmpegCommandsJSON() (string, error) { return marshalJSONString(commands) } +// WaitForPendingFFmpegCommandsJSON blocks until work is available or the +// timeout elapses. Native command pumps use this instead of polling the bridge +// every 100 ms while still retaining GetAllPendingFFmpegCommandsJSON for older +// clients. +func WaitForPendingFFmpegCommandsJSON(timeoutMillis int64) (string, error) { + if timeoutMillis < 0 { + timeoutMillis = 0 + } + deadline := time.NewTimer(time.Duration(timeoutMillis) * time.Millisecond) + defer deadline.Stop() + + for { + ffmpegCommandsMu.Lock() + commands := make([]map[string]any, 0) + for cmdID, cmd := range ffmpegCommands { + if cmd.Completed || cmd.Claimed { + continue + } + cmd.Claimed = true + commands = append(commands, map[string]any{ + "command_id": cmdID, + "extension_id": cmd.ExtensionID, + "command": cmd.Command, + }) + } + ffmpegCommandsMu.Unlock() + if len(commands) > 0 { + return marshalJSONString(commands) + } + + select { + case <-ffmpegCommandQueued: + // A buffered notification can be stale if another command pump + // already consumed the work, so re-check until the deadline. + continue + case <-deadline.C: + return "[]", nil + } + } +} + func EnrichTrackWithExtensionJSON(extensionID, trackJSON string) (string, error) { manager := getExtensionManager() ext, err := manager.GetExtension(extensionID) diff --git a/go_backend/extension_runtime_chunked.go b/go_backend/extension_runtime_chunked.go index 852ad362..49b5091c 100644 --- a/go_backend/extension_runtime_chunked.go +++ b/go_backend/extension_runtime_chunked.go @@ -211,6 +211,7 @@ func (r *extensionRuntime) fileDownloadChunked( } shouldTrackBytes := activeItemID != "" && trackItemBytes + itemProgressReporter := NewItemTransferProgressReporter(activeItemID, totalWritten, totalSize) if shouldTrackBytes { if totalSize > 0 { SetItemProgress( @@ -413,16 +414,7 @@ func (r *extensionRuntime) fileDownloadChunked( }) } if shouldTrackBytes { - if totalSize > 0 { - SetItemProgress( - activeItemID, - float64(totalWritten)/float64(totalSize), - totalWritten, - totalSize, - ) - } else { - SetItemBytesReceived(activeItemID, totalWritten) - } + itemProgressReporter.Report(totalWritten, totalSize) } if onProgress != nil && totalSize > 0 && (totalWritten-lastProgressNotify >= progressUpdateThreshold || totalWritten >= totalSize) { diff --git a/go_backend/extension_runtime_ffmpeg.go b/go_backend/extension_runtime_ffmpeg.go index 21d225ef..c2f906c1 100644 --- a/go_backend/extension_runtime_ffmpeg.go +++ b/go_backend/extension_runtime_ffmpeg.go @@ -16,17 +16,27 @@ type FFmpegCommand struct { InputPath string OutputPath string Completed bool + Claimed bool Success bool Error string Output string + done chan struct{} } var ( - ffmpegCommands = make(map[string]*FFmpegCommand) - ffmpegCommandsMu sync.RWMutex - ffmpegCommandID int64 + ffmpegCommands = make(map[string]*FFmpegCommand) + ffmpegCommandsMu sync.RWMutex + ffmpegCommandID int64 + ffmpegCommandQueued = make(chan struct{}, 1) ) +func notifyFFmpegCommandQueued() { + select { + case ffmpegCommandQueued <- struct{}{}: + default: + } +} + func GetPendingFFmpegCommand(commandID string) *FFmpegCommand { ffmpegCommandsMu.RLock() defer ffmpegCommandsMu.RUnlock() @@ -37,10 +47,16 @@ func SetFFmpegCommandResult(commandID string, success bool, output, errorMsg str ffmpegCommandsMu.Lock() defer ffmpegCommandsMu.Unlock() if cmd, exists := ffmpegCommands[commandID]; exists { + if cmd.Completed { + return + } cmd.Completed = true cmd.Success = success cmd.Output = output cmd.Error = errorMsg + if cmd.done != nil { + close(cmd.done) + } } } @@ -66,46 +82,36 @@ func (r *extensionRuntime) executeFFmpegCommand(command, inputPath, outputPath s ffmpegCommandsMu.Lock() ffmpegCommandID++ cmdID := fmt.Sprintf("%s_%d", r.extensionID, ffmpegCommandID) - ffmpegCommands[cmdID] = &FFmpegCommand{ + queuedCommand := &FFmpegCommand{ ExtensionID: r.extensionID, Command: command, InputPath: inputPath, OutputPath: outputPath, Completed: false, + done: make(chan struct{}), } + ffmpegCommands[cmdID] = queuedCommand ffmpegCommandsMu.Unlock() + notifyFFmpegCommandQueued() GoLog("[Extension:%s] FFmpeg command queued: %s\n", r.extensionID, cmdID) - timeout := 5 * time.Minute - start := time.Now() - for { - ffmpegCommandsMu.RLock() - cmd := ffmpegCommands[cmdID] - completed := cmd != nil && cmd.Completed - ffmpegCommandsMu.RUnlock() - - if completed { - ffmpegCommandsMu.RLock() - result := map[string]any{ - "success": cmd.Success, - "output": cmd.Output, - } - if cmd.Error != "" { - result["error"] = cmd.Error - } - ffmpegCommandsMu.RUnlock() - - ClearFFmpegCommand(cmdID) - return r.vm.ToValue(result) + select { + case <-queuedCommand.done: + ffmpegCommandsMu.Lock() + result := map[string]any{ + "success": queuedCommand.Success, + "output": queuedCommand.Output, } - - if time.Since(start) > timeout { - ClearFFmpegCommand(cmdID) - return r.jsError("FFmpeg command timed out") + if queuedCommand.Error != "" { + result["error"] = queuedCommand.Error } - - time.Sleep(100 * time.Millisecond) + delete(ffmpegCommands, cmdID) + ffmpegCommandsMu.Unlock() + return r.vm.ToValue(result) + case <-time.After(5 * time.Minute): + ClearFFmpegCommand(cmdID) + return r.jsError("FFmpeg command timed out") } } diff --git a/go_backend/extension_runtime_ffmpeg_test.go b/go_backend/extension_runtime_ffmpeg_test.go new file mode 100644 index 00000000..42ae6fca --- /dev/null +++ b/go_backend/extension_runtime_ffmpeg_test.go @@ -0,0 +1,36 @@ +package gobackend + +import ( + "strings" + "testing" +) + +func TestWaitForPendingFFmpegCommandsClaimsCommandOnce(t *testing.T) { + const commandID = "wait-claim-test" + command := &FFmpegCommand{ + ExtensionID: "test-extension", + Command: "ffmpeg -version", + done: make(chan struct{}), + } + ffmpegCommandsMu.Lock() + ffmpegCommands[commandID] = command + ffmpegCommandsMu.Unlock() + notifyFFmpegCommandQueued() + t.Cleanup(func() { ClearFFmpegCommand(commandID) }) + + first, err := WaitForPendingFFmpegCommandsJSON(50) + if err != nil || !strings.Contains(first, commandID) { + t.Fatalf("first wait = %q, %v", first, err) + } + second, err := WaitForPendingFFmpegCommandsJSON(1) + if err != nil || second != "[]" { + t.Fatalf("claimed command returned twice: %q, %v", second, err) + } + + SetFFmpegCommandResult(commandID, true, "ok", "") + select { + case <-command.done: + default: + t.Fatal("command completion did not signal waiter") + } +} diff --git a/go_backend/extension_runtime_segments.go b/go_backend/extension_runtime_segments.go index 12e52cca..f5125dec 100644 --- a/go_backend/extension_runtime_segments.go +++ b/go_backend/extension_runtime_segments.go @@ -169,7 +169,7 @@ func (r *extensionRuntime) fetchSegmentToTemp( tempPath string, policy DownloadTransferPolicy, received *atomic.Int64, - activeItemID string, + itemProgressReporter *ItemTransferProgressReporter, ) segmentTransferResult { config := transferRetryConfig(policy) retryDelay := config.InitialDelay @@ -291,9 +291,7 @@ func (r *extensionRuntime) fetchSegmentToTemp( writeCount, writeErr := output.Write(buffer[:readCount]) size += int64(writeCount) received.Add(int64(writeCount)) - if activeItemID != "" { - SetItemBytesReceived(activeItemID, received.Load()) - } + itemProgressReporter.Report(received.Load(), 0) if writeErr != nil || writeCount != readCount { if writeErr == nil { writeErr = io.ErrShortWrite @@ -564,6 +562,7 @@ func (r *extensionRuntime) fileDownloadSegments(call goja.FunctionCall) goja.Val results := make(chan segmentTransferResult, policy.MaxParallelSegments) var received atomic.Int64 received.Store(totalWritten) + itemProgressReporter := NewItemTransferProgressReporter(activeItemID, totalWritten, 0) var workers sync.WaitGroup workerCount := min(policy.MaxParallelSegments, len(segments)-nextIndex) for workerIndex := 0; workerIndex < workerCount; workerIndex++ { @@ -578,7 +577,7 @@ func (r *extensionRuntime) fileDownloadSegments(call goja.FunctionCall) goja.Val segmentTempPath(stagedPath, spec.Index), policy, &received, - activeItemID, + itemProgressReporter, ) select { case results <- result: diff --git a/go_backend/extension_runtime_storage.go b/go_backend/extension_runtime_storage.go index 390166e3..06b4d927 100644 --- a/go_backend/extension_runtime_storage.go +++ b/go_backend/extension_runtime_storage.go @@ -20,6 +20,22 @@ import ( // process-wide; the per-runtime mutexes only cover a single VM. var extensionFileMus sync.Map // file path -> *sync.Mutex +type extensionFileIdentity struct { + exists bool + size int64 + modified int64 +} + +type extensionJSONCacheEntry struct { + identity extensionFileIdentity + snapshot map[string]any +} + +// Shared by all isolated runtimes so repeated storage/credential reads avoid +// reading, decoding, and (for credentials) decrypting the complete file. The +// corresponding extensionFileMu must be held while accessing an entry. +var extensionJSONCaches sync.Map // file path -> *extensionJSONCacheEntry + func extensionFileMu(path string) *sync.Mutex { mu, _ := extensionFileMus.LoadOrStore(path, &sync.Mutex{}) return mu.(*sync.Mutex) @@ -35,6 +51,86 @@ func writeExtensionFileLocked(path string, data []byte) error { return os.Rename(tmp, path) } +func extensionFileIdentityForPath(path string) (extensionFileIdentity, error) { + info, err := os.Stat(path) + if err != nil { + if os.IsNotExist(err) { + return extensionFileIdentity{}, nil + } + return extensionFileIdentity{}, err + } + return extensionFileIdentity{ + exists: true, + size: info.Size(), + modified: info.ModTime().UnixNano(), + }, nil +} + +func cloneJSONValue(value any) any { + switch typed := value.(type) { + case map[string]any: + return cloneJSONMap(typed) + case []any: + result := make([]any, len(typed)) + for index, item := range typed { + result[index] = cloneJSONValue(item) + } + return result + default: + return typed + } +} + +func cloneJSONMap(source map[string]any) map[string]any { + result := make(map[string]any, len(source)) + for key, value := range source { + result[key] = cloneJSONValue(value) + } + return result +} + +// readCachedJSONMapLocked returns an isolated snapshot. The path-specific +// mutex must be held, which makes the stat/load/update sequence coherent with +// writers from all runtimes. +func readCachedJSONMapLocked( + path string, + load func() (map[string]any, error), +) (map[string]any, error) { + identity, err := extensionFileIdentityForPath(path) + if err != nil { + return nil, err + } + if cached, ok := extensionJSONCaches.Load(path); ok { + entry := cached.(*extensionJSONCacheEntry) + if entry.identity == identity { + return cloneJSONMap(entry.snapshot), nil + } + } + + snapshot, err := load() + if err != nil { + return nil, err + } + extensionJSONCaches.Store(path, &extensionJSONCacheEntry{ + identity: identity, + snapshot: cloneJSONMap(snapshot), + }) + return snapshot, nil +} + +func storeCachedJSONMapLocked(path string, snapshot map[string]any) error { + identity, err := extensionFileIdentityForPath(path) + if err != nil { + extensionJSONCaches.Delete(path) + return err + } + extensionJSONCaches.Store(path, &extensionJSONCacheEntry{ + identity: identity, + snapshot: cloneJSONMap(snapshot), + }) + return nil +} + func (r *extensionRuntime) getStoragePath() string { return filepath.Join(r.dataDir, "storage.json") } @@ -61,7 +157,9 @@ func (r *extensionRuntime) refreshStorage() error { path := r.getStoragePath() fileMu := extensionFileMu(path) fileMu.Lock() - snapshot, err := readJSONMapFile(path) + snapshot, err := readCachedJSONMapLocked(path, func() (map[string]any, error) { + return readJSONMapFile(path) + }) fileMu.Unlock() if err != nil { return err @@ -83,13 +181,18 @@ func (r *extensionRuntime) mutateStorage(mutate func(map[string]any) bool) error path := r.getStoragePath() fileMu := extensionFileMu(path) fileMu.Lock() - snapshot, err := readJSONMapFile(path) + snapshot, err := readCachedJSONMapLocked(path, func() (map[string]any, error) { + return readJSONMapFile(path) + }) if err == nil && mutate(snapshot) { var data []byte data, err = json.Marshal(snapshot) if err == nil { err = writeExtensionFileLocked(path, data) } + if err == nil { + err = storeCachedJSONMapLocked(path, snapshot) + } } fileMu.Unlock() if err != nil { @@ -253,7 +356,7 @@ func (r *extensionRuntime) refreshCredentials() error { path := r.getCredentialsPath() fileMu := extensionFileMu(path) fileMu.Lock() - snapshot, err := r.readCredentialsFileLocked() + snapshot, err := readCachedJSONMapLocked(path, r.readCredentialsFileLocked) fileMu.Unlock() if err != nil { return err @@ -268,7 +371,7 @@ func (r *extensionRuntime) mutateCredentials(mutate func(map[string]any)) error path := r.getCredentialsPath() fileMu := extensionFileMu(path) fileMu.Lock() - snapshot, err := r.readCredentialsFileLocked() + snapshot, err := readCachedJSONMapLocked(path, r.readCredentialsFileLocked) if err == nil { mutate(snapshot) var data []byte @@ -283,6 +386,9 @@ func (r *extensionRuntime) mutateCredentials(mutate func(map[string]any)) error if err == nil { err = writeExtensionFileLocked(path, data) } + if err == nil { + err = storeCachedJSONMapLocked(path, snapshot) + } } fileMu.Unlock() if err != nil { diff --git a/go_backend/extension_runtime_storage_test.go b/go_backend/extension_runtime_storage_test.go index dfa61164..9a6890c7 100644 --- a/go_backend/extension_runtime_storage_test.go +++ b/go_backend/extension_runtime_storage_test.go @@ -24,6 +24,50 @@ func setStorageValue(t *testing.T, runtime *extensionRuntime, key string, value } } +func TestExtensionJSONCacheUsesIdentityAndIsolatesSnapshots(t *testing.T) { + path := filepath.Join(t.TempDir(), "storage.json") + if err := os.WriteFile(path, []byte(`{"value":"first"}`), 0600); err != nil { + t.Fatal(err) + } + loads := 0 + load := func() (map[string]any, error) { + loads++ + return readJSONMapFile(path) + } + + mu := extensionFileMu(path) + mu.Lock() + first, err := readCachedJSONMapLocked(path, load) + mu.Unlock() + if err != nil { + t.Fatal(err) + } + first["value"] = "mutated locally" + + mu.Lock() + second, err := readCachedJSONMapLocked(path, load) + mu.Unlock() + if err != nil { + t.Fatal(err) + } + if loads != 1 || second["value"] != "first" { + t.Fatalf("cache did not isolate/reuse snapshot: loads=%d data=%#v", loads, second) + } + + if err := os.WriteFile(path, []byte(`{"value":"external update"}`), 0600); err != nil { + t.Fatal(err) + } + mu.Lock() + third, err := readCachedJSONMapLocked(path, load) + mu.Unlock() + if err != nil { + t.Fatal(err) + } + if loads != 2 || third["value"] != "external update" { + t.Fatalf("external update was not reloaded: loads=%d data=%#v", loads, third) + } +} + func TestExtensionRuntimeStorageConcurrentRuntimesMergeWrites(t *testing.T) { dataDir := t.TempDir() ext := &loadedExtension{ID: "merge-test", Manifest: &ExtensionManifest{Name: "merge-test"}, DataDir: dataDir} diff --git a/go_backend/extension_runtime_transfer.go b/go_backend/extension_runtime_transfer.go index 02b5957c..abda6eec 100644 --- a/go_backend/extension_runtime_transfer.go +++ b/go_backend/extension_runtime_transfer.go @@ -293,6 +293,7 @@ func (r *extensionRuntime) reliableFileDownload( SetItemDownloading(activeItemID) } shouldTrackItemBytes := activeItemID != "" && trackItemBytes + itemProgressReporter := NewItemTransferProgressReporter(activeItemID, written, contentLength) if shouldTrackItemBytes { if contentLength > 0 { SetItemProgress(activeItemID, float64(written)/float64(contentLength), written, contentLength) @@ -541,12 +542,7 @@ func (r *extensionRuntime) reliableFileDownload( contentLength = resp.ContentLength } if shouldTrackItemBytes && contentLength > 0 { - SetItemProgress( - activeItemID, - float64(written)/float64(contentLength), - written, - contentLength, - ) + itemProgressReporter.Report(written, contentLength) } var readErr error buffer := make([]byte, 64*1024) @@ -569,11 +565,7 @@ func (r *extensionRuntime) reliableFileDownload( }) } if shouldTrackItemBytes { - if contentLength > 0 { - SetItemProgress(activeItemID, float64(written)/float64(contentLength), written, contentLength) - } else { - SetItemBytesReceived(activeItemID, written) - } + itemProgressReporter.Report(written, contentLength) } if onProgress != nil && contentLength > 0 && (written-lastProgressNotify >= progressUpdateThreshold || written >= contentLength) { diff --git a/go_backend/progress.go b/go_backend/progress.go index 2d8b81ee..33b5a4b6 100644 --- a/go_backend/progress.go +++ b/go_backend/progress.go @@ -407,6 +407,54 @@ type ItemProgressWriter struct { } const progressUpdateThreshold = 128 * 1024 +const progressUpdateMaxInterval = 250 * time.Millisecond + +// ItemTransferProgressReporter coalesces hot-path byte updates before they +// acquire multiMu. Transfer loops commonly read in 64 KiB chunks; reporting +// every read needlessly serializes parallel workers even though the bridge +// exposes progress at a much lower cadence. +type ItemTransferProgressReporter struct { + itemID string + mu sync.Mutex + lastReported int64 + lastTotal int64 + lastReportAt time.Time +} + +func NewItemTransferProgressReporter(itemID string, received, total int64) *ItemTransferProgressReporter { + return &ItemTransferProgressReporter{ + itemID: itemID, + lastReported: received, + lastTotal: total, + lastReportAt: time.Now(), + } +} + +func (reporter *ItemTransferProgressReporter) Report(received, total int64) { + if reporter == nil || reporter.itemID == "" { + return + } + + reporter.mu.Lock() + defer reporter.mu.Unlock() + now := time.Now() + bytesDelta := received - reporter.lastReported + if bytesDelta >= 0 && + bytesDelta < progressUpdateThreshold && + total == reporter.lastTotal && + now.Sub(reporter.lastReportAt) < progressUpdateMaxInterval { + return + } + + reporter.lastReported = received + reporter.lastTotal = total + reporter.lastReportAt = now + if total > 0 { + SetItemProgress(reporter.itemID, float64(received)/float64(total), received, total) + } else { + SetItemBytesReceived(reporter.itemID, received) + } +} func NewItemProgressWriter(w interface{ Write([]byte) (int, error) }, itemID string) *ItemProgressWriter { now := time.Now() diff --git a/go_backend/progress_test.go b/go_backend/progress_test.go index 38166d20..3fd64c29 100644 --- a/go_backend/progress_test.go +++ b/go_backend/progress_test.go @@ -3,8 +3,37 @@ package gobackend import ( "encoding/json" "testing" + "time" ) +func TestItemTransferProgressReporterCoalescesHotPathUpdates(t *testing.T) { + ClearAllItemProgress() + defer ClearAllItemProgress() + + const itemID = "coalesced-transfer-progress" + const total = int64(1024 * 1024) + StartItemProgress(itemID) + SetItemDownloading(itemID) + SetItemBytesTotal(itemID, total) + reporter := NewItemTransferProgressReporter(itemID, 0, total) + + reporter.Report(64*1024, total) + if received := multiProgress.Items[itemID].BytesReceived; received != 0 { + t.Fatalf("sub-threshold bytes = %d, want 0", received) + } + + reporter.Report(progressUpdateThreshold, total) + if received := multiProgress.Items[itemID].BytesReceived; received != progressUpdateThreshold { + t.Fatalf("threshold bytes = %d, want %d", received, progressUpdateThreshold) + } + + reporter.lastReportAt = time.Now().Add(-progressUpdateMaxInterval) + reporter.Report(progressUpdateThreshold+1, total) + if received := multiProgress.Items[itemID].BytesReceived; received != progressUpdateThreshold+1 { + t.Fatalf("interval flush bytes = %d, want %d", received, progressUpdateThreshold+1) + } +} + func TestItemProgressPreparingAndDownloadingStatuses(t *testing.T) { const itemID = "progress-phase-item" RemoveItemProgress(itemID) diff --git a/lib/providers/local_library_provider.dart b/lib/providers/local_library_provider.dart index be841377..33645c13 100644 --- a/lib/providers/local_library_provider.dart +++ b/lib/providers/local_library_provider.dart @@ -3,15 +3,12 @@ import 'dart:io'; import 'package:flutter_riverpod/flutter_riverpod.dart'; import 'package:path_provider/path_provider.dart'; import 'package:shared_preferences/shared_preferences.dart'; -import 'package:spotiflac_android/providers/download_queue_provider.dart'; import 'package:spotiflac_android/providers/settings_provider.dart'; -import 'package:spotiflac_android/services/history_database.dart'; import 'package:spotiflac_android/services/library_database.dart'; import 'package:spotiflac_android/services/notification_service.dart'; import 'package:spotiflac_android/services/platform_bridge.dart'; import 'package:spotiflac_android/utils/logger.dart'; import 'package:spotiflac_android/utils/local_library_scan_prefs.dart'; -import 'package:spotiflac_android/utils/path_match_keys.dart'; import 'package:spotiflac_android/utils/progress_stream_poller.dart'; final _log = AppLogger('LocalLibrary'); @@ -122,7 +119,6 @@ class LocalLibraryState { class LocalLibraryNotifier extends Notifier { final LibraryDatabase _db = LibraryDatabase.instance; - final HistoryDatabase _historyDb = HistoryDatabase.instance; final NotificationService _notificationService = NotificationService(); static const _progressPollingInterval = Duration(milliseconds: 350); static const _progressStreamBootstrapTimeout = Duration(milliseconds: 900); @@ -520,24 +516,10 @@ class LocalLibraryNotifier extends Notifier { return reconnected; } - bool _isDownloadedPath(String? filePath, Set downloadedPathKeys) { - if (filePath == null || filePath.isEmpty || downloadedPathKeys.isEmpty) { - return false; - } - final candidateKeys = buildPathMatchKeys(filePath); - for (final key in candidateKeys) { - if (downloadedPathKeys.contains(key)) { - return true; - } - } - return false; - } - Future<({int inserted, int skipped})?> _replaceFromFullScanStream({ required String sourceId, required String folderPath, required bool isSaf, - required Set downloadedPathKeys, }) async { if (_scanCancelRequested) return null; final scanFile = isSaf @@ -549,7 +531,6 @@ class LocalLibraryNotifier extends Notifier { folderPath, isCancelled: () => _scanCancelRequested, ); - var skipped = 0; try { if (_scanCancelRequested) return null; state = state.copyWith( @@ -557,18 +538,13 @@ class LocalLibraryNotifier extends Notifier { scanProgress: state.scanProgress >= 99 ? state.scanProgress : 99, scanCurrentFile: null, ); - Stream> filteredRows() async* { + Stream> validatedRows() async* { var decodedRows = 0; await for (final json in scanFile.rows()) { if (_scanCancelRequested) { throw StateError('Library scan cancelled during ingestion'); } decodedRows++; - final filePath = json['filePath'] as String?; - if (_isDownloadedPath(filePath, downloadedPathKeys)) { - skipped++; - continue; - } yield json; } if (decodedRows != scanFile.expectedCount) { @@ -579,12 +555,12 @@ class LocalLibraryNotifier extends Notifier { } } - final inserted = await _db.replaceSourceStream(sourceId, filteredRows()); + final result = await _db.replaceSourceStream(sourceId, validatedRows()); _log.i( - 'Stream-ingested $inserted/${scanFile.expectedCount} scan rows ' - '($skipped downloads excluded)', + 'Stream-ingested ${result.inserted}/${scanFile.expectedCount} scan rows ' + '(${result.skipped} downloads excluded)', ); - return (inserted: inserted, skipped: skipped); + return result; } finally { await scanFile.delete(); } @@ -684,25 +660,6 @@ class LocalLibraryNotifier extends Notifier { try { final isSaf = effectiveFolderPath.startsWith('content://'); - final downloadedPaths = await _historyDb.getAllFilePaths(); - final inMemoryHistoryPaths = ref - .read(downloadHistoryProvider) - .items - .map((item) => item.filePath) - .where((path) => path.isNotEmpty); - final allHistoryPaths = { - ...downloadedPaths, - ...inMemoryHistoryPaths, - }; - final downloadedPathKeys = {}; - for (final path in allHistoryPaths) { - downloadedPathKeys.addAll(buildPathMatchKeys(path)); - } - _log.i( - 'Excluding ${allHistoryPaths.length} downloaded files from library scan ' - '(${downloadedPathKeys.length} path keys)', - ); - final useStreamingFullScan = forceFullScan || await _db.getSourceCount(activeSourceId) == 0; if (useStreamingFullScan) { @@ -710,7 +667,6 @@ class LocalLibraryNotifier extends Notifier { sourceId: activeSourceId, folderPath: effectiveFolderPath, isSaf: isSaf, - downloadedPathKeys: downloadedPathKeys, ); if (scanResult == null || _scanCancelRequested) { state = state.copyWith( @@ -868,39 +824,31 @@ class LocalLibraryNotifier extends Notifier { '$skippedCount skipped, ${deletedPaths.length} deleted, $totalFiles total', ); - final existingPaths = existingFiles.keys.toList(growable: false); - final existingDownloadedPaths = []; - for (final path in existingPaths) { - if (_isDownloadedPath(path, downloadedPathKeys)) { - existingDownloadedPaths.add(path); - } - } - if (existingDownloadedPaths.isNotEmpty) { - final removed = await _db.deleteByPaths(existingDownloadedPaths); + final removedDownloaded = await _db.deleteDownloadedRowsForSource( + activeSourceId, + ); + if (removedDownloaded > 0) { _log.i( - 'Removed $removed downloaded tracks already present in local library index', + 'Removed $removedDownloaded downloaded tracks already present in ' + 'the local Library index', ); } final updatedItems = []; - int skippedDownloads = existingDownloadedPaths.length; + var skippedDownloads = removedDownloaded; if (scannedList.isNotEmpty) { for (final json in scannedList) { final map = json as Map; - final filePath = map['filePath'] as String?; - if (_isDownloadedPath(filePath, downloadedPathKeys)) { - skippedDownloads++; - continue; - } final item = LocalLibraryItem.fromJson(map); updatedItems.add(item); } if (updatedItems.isNotEmpty) { - await _db.upsertBatch( + final upsertResult = await _db.upsertBatchExcludingHistory( updatedItems.map((e) => e.toJson()).toList(), sourceId: activeSourceId, ); - _log.i('Upserted ${updatedItems.length} items'); + skippedDownloads += upsertResult.skipped; + _log.i('Upserted ${upsertResult.upserted} items'); } if (skippedDownloads > 0) { _log.i( diff --git a/lib/screens/downloaded_album_screen.dart b/lib/screens/downloaded_album_screen.dart index 99e49df2..86f2abbf 100644 --- a/lib/screens/downloaded_album_screen.dart +++ b/lib/screens/downloaded_album_screen.dart @@ -554,38 +554,47 @@ class _DownloadedAlbumScreenState extends ConsumerState } final discNumbers = _getSortedDiscNumbers(tracks); - final List children = []; + final navigationIndexById = { + for (var index = 0; index < tracks.length; index++) + tracks[index].id: index, + }; + final slivers = []; var revealIndex = 0; for (final discNumber in discNumbers) { final discTracks = discMap[discNumber]; if (discTracks == null || discTracks.isEmpty) continue; - children.add(DiscSeparatorChip(discNumber: discNumber)); - - for (final track in discTracks) { - final navigationIndex = tracks.indexOf(track); - children.add( - KeyedSubtree( - key: ValueKey(track.id), - child: StaggeredListItem( - index: revealIndex++, - child: _buildTrackItem( - context, - colorScheme, - track, - tracks, - navigationIndex, + slivers.add( + SliverToBoxAdapter(child: DiscSeparatorChip(discNumber: discNumber)), + ); + final discRevealStart = revealIndex; + revealIndex += discTracks.length; + slivers.add( + SliverList( + delegate: SliverChildBuilderDelegate((context, index) { + final track = discTracks[index]; + return KeyedSubtree( + key: ValueKey(track.id), + child: StaggeredListItem( + index: discRevealStart + index, + child: _buildTrackItem( + context, + colorScheme, + track, + tracks, + navigationIndexById[track.id] ?? 0, + ), ), - ), - ), - ); - } + ); + }, childCount: discTracks.length), + ), + ); } return SliverPadding( padding: EdgeInsets.symmetric(horizontal: wideListInset(context)), - sliver: SliverList(delegate: SliverChildListDelegate(children)), + sliver: SliverMainAxisGroup(slivers: slivers), ); } diff --git a/lib/services/ffmpeg_service.dart b/lib/services/ffmpeg_service.dart index cb237160..44271df9 100644 --- a/lib/services/ffmpeg_service.dart +++ b/lib/services/ffmpeg_service.dart @@ -31,6 +31,21 @@ class _ResolvedLosslessConversionQuality { }); } +class _PrimaryAudioProperties { + final String? codec; + final int? bitDepth; + final int? sampleRate; + + const _PrimaryAudioProperties({this.codec, this.bitDepth, this.sampleRate}); +} + +class _AudioProbeCacheEntry { + final String identity; + final Future<_PrimaryAudioProperties> result; + + const _AudioProbeCacheEntry({required this.identity, required this.result}); +} + class _ConversionOutputPlan { final String workingPath; final String finalPath; @@ -45,6 +60,7 @@ class _ConversionOutputPlan { class FFmpegService { static const int _commandLogPreviewLength = 300; + static const int _audioProbeCacheMaxEntries = 64; static const Duration _liveTunnelStartupTimeout = Duration(seconds: 8); static const Duration _liveTunnelStartupPollInterval = Duration( milliseconds: 200, @@ -60,6 +76,8 @@ class FFmpegService { static String? _activeNativeDashManifestPath; static String? _activeNativeDashManifestUrl; static final Set _preparedNativeDashManifestPaths = {}; + static final Map _audioProbeCache = + {}; static String _buildOutputPath(String inputPath, String extension) { final normalizedExt = extension.startsWith('.') ? extension : '.$extension'; @@ -407,21 +425,71 @@ class FFmpegService { } static Future probePrimaryAudioCodec(String filePath) async { + return (await _probePrimaryAudioProperties(filePath)).codec; + } + + static Future _audioProbeIdentity(String filePath) async { + try { + final stat = await File(filePath).stat(); + if (stat.type != FileSystemEntityType.file) return null; + return '${stat.modified.millisecondsSinceEpoch}:${stat.size}'; + } catch (_) { + return null; + } + } + + static Future<_PrimaryAudioProperties> _probePrimaryAudioProperties( + String filePath, + ) async { + final identity = await _audioProbeIdentity(filePath); + if (identity != null) { + final cached = _audioProbeCache[filePath]; + if (cached != null && cached.identity == identity) { + return cached.result; + } + } + + final result = _readPrimaryAudioProperties(filePath); + if (identity != null) { + while (_audioProbeCache.length >= _audioProbeCacheMaxEntries && + _audioProbeCache.isNotEmpty) { + _audioProbeCache.remove(_audioProbeCache.keys.first); + } + _audioProbeCache[filePath] = _AudioProbeCacheEntry( + identity: identity, + result: result, + ); + } + return result; + } + + static Future<_PrimaryAudioProperties> _readPrimaryAudioProperties( + String filePath, + ) async { try { final session = await FFprobeKit.getMediaInformation(filePath); final info = session.getMediaInformation(); - if (info == null) return null; + if (info == null) return const _PrimaryAudioProperties(); for (final stream in info.getStreams()) { final props = stream.getAllProperties() ?? const {}; if (props['codec_type']?.toString() != 'audio') continue; final codec = props['codec_name']?.toString().trim().toLowerCase(); - return codec == null || codec.isEmpty ? null : codec; + final rawBits = props['bits_per_raw_sample']?.toString(); + final bits = props['bits_per_sample']?.toString(); + final bitDepth = + int.tryParse(rawBits ?? '') ?? int.tryParse(bits ?? ''); + final sampleRate = int.tryParse(props['sample_rate']?.toString() ?? ''); + return _PrimaryAudioProperties( + codec: codec == null || codec.isEmpty ? null : codec, + bitDepth: bitDepth != null && bitDepth > 0 ? bitDepth : null, + sampleRate: sampleRate != null && sampleRate > 0 ? sampleRate : null, + ); } } catch (e) { - _log.w('Audio codec probe failed for $filePath: $e'); + _log.w('Audio property probe failed for $filePath: $e'); } - return null; + return const _PrimaryAudioProperties(); } static bool isLosslessAudioCodec(String? codec) { @@ -443,41 +511,11 @@ class FFmpegService { /// Probes the source audio bit depth (bits_per_raw_sample, falling back to /// bits_per_sample). Returns null when unknown. static Future probeBitDepth(String filePath) async { - try { - final session = await FFprobeKit.getMediaInformation(filePath); - final info = session.getMediaInformation(); - if (info == null) return null; - for (final stream in info.getStreams()) { - final props = stream.getAllProperties() ?? const {}; - if (props['codec_type']?.toString() != 'audio') continue; - final raw = props['bits_per_raw_sample']?.toString(); - final bps = props['bits_per_sample']?.toString(); - final v = int.tryParse(raw ?? '') ?? int.tryParse(bps ?? ''); - if (v != null && v > 0) return v; - return null; - } - } catch (e) { - _log.w('Bit depth probe failed for $filePath: $e'); - } - return null; + return (await _probePrimaryAudioProperties(filePath)).bitDepth; } static Future probeSampleRate(String filePath) async { - try { - final session = await FFprobeKit.getMediaInformation(filePath); - final info = session.getMediaInformation(); - if (info == null) return null; - for (final stream in info.getStreams()) { - final props = stream.getAllProperties() ?? const {}; - if (props['codec_type']?.toString() != 'audio') continue; - final value = int.tryParse(props['sample_rate']?.toString() ?? ''); - if (value != null && value > 0) return value; - return null; - } - } catch (e) { - _log.w('Sample rate probe failed for $filePath: $e'); - } - return null; + return (await _probePrimaryAudioProperties(filePath)).sampleRate; } /// Returns `true` when [filePath] starts with the native FLAC magic bytes @@ -507,12 +545,14 @@ class FFmpegService { required LosslessConversionQuality quality, int? sourceBitDepth, }) async { - final probedBitDepth = - sourceBitDepth ?? - (quality.maxBitDepth != null ? await probeBitDepth(inputPath) : null); - final probedSampleRate = quality.maxSampleRate != null - ? await probeSampleRate(inputPath) - : null; + final needsBitDepthProbe = + sourceBitDepth == null && quality.maxBitDepth != null; + final needsSampleRateProbe = quality.maxSampleRate != null; + final probe = needsBitDepthProbe || needsSampleRateProbe + ? await _probePrimaryAudioProperties(inputPath) + : const _PrimaryAudioProperties(); + final probedBitDepth = sourceBitDepth ?? probe.bitDepth; + final probedSampleRate = needsSampleRateProbe ? probe.sampleRate : null; int? targetBitDepth; if (quality.maxBitDepth != null && diff --git a/lib/services/history_database.dart b/lib/services/history_database.dart index b11041fe..cd23e080 100644 --- a/lib/services/history_database.dart +++ b/lib/services/history_database.dart @@ -928,72 +928,118 @@ class HistoryDatabase { return null; } - /// Batch variant used by playlist playback. Four indexed scans resolve the - /// complete list while preserving the same Spotify -> ISRC -> match-key - /// priority as [findExistingTrack]. + /// Batch variant used by playlist playback and bulk existence checks. A + /// compact candidates CTE resolves all identifiers in one indexed lookup per + /// bounded chunk, returning only the id/path reference those callers need. Future?>> findExistingTracks( List requests, ) async { if (requests.isEmpty) return const []; final db = await database; - final bySpotify = >{}; - final bySpotifyNorm = >{}; - final byIsrcNorm = >{}; - final byMatchKey = >{}; + final results = List?>.filled(requests.length, null); + const requestChunkSize = 80; - Future loadColumn( - String column, - Iterable rawValues, - Map> destination, + for ( + var chunkStart = 0; + chunkStart < requests.length; + chunkStart += requestChunkSize ) { - return sqlite.loadRowsByColumn( - db, - table: 'history', - column: column, - rawValues: rawValues, - destination: destination, - mapRow: _dbRowToJson, - orderBy: 'sort_added DESC, id DESC', + final chunkEnd = (chunkStart + requestChunkSize).clamp( + 0, + requests.length, ); - } + final chunk = requests.sublist(chunkStart, chunkEnd); + final candidateRows = []; + final args = []; - final spotifyCandidates = requests.expand( - (request) => spotifyLookupCandidates(request.spotifyId), - ); - await Future.wait([ - loadColumn('spotify_id', spotifyCandidates, bySpotify), - loadColumn( - 'spotify_id_norm', - spotifyCandidates.map(normalizeSpotifyId), - bySpotifyNorm, - ), - loadColumn( - 'isrc_norm', - requests.map((request) => normalizeIsrc(request.isrc)), - byIsrcNorm, - ), - loadColumn( - 'match_key', - requests.map( - (request) => matchKeyFor(request.trackName, request.artistName), - ), - byMatchKey, - ), - ]); + void addCandidate( + int requestIndex, + int priority, + String kind, + String value, + ) { + if (value.isEmpty) return; + candidateRows.add("($requestIndex, $priority, '$kind', ?)"); + args.add(value); + } - return requests - .map((request) { - for (final candidate in spotifyLookupCandidates(request.spotifyId)) { - final match = - bySpotify[candidate] ?? - bySpotifyNorm[normalizeSpotifyId(candidate)]; - if (match != null) return match; + for (var requestIndex = 0; requestIndex < chunk.length; requestIndex++) { + final request = chunk[requestIndex]; + var priority = 0; + final seen = {}; + for (final candidate in spotifyLookupCandidates(request.spotifyId)) { + final exactKey = 'spotify_id\u0000$candidate'; + if (candidate.isNotEmpty && seen.add(exactKey)) { + addCandidate(requestIndex, priority++, 'spotify_id', candidate); } - final byIsrc = byIsrcNorm[normalizeIsrc(request.isrc)]; - if (byIsrc != null) return byIsrc; - return byMatchKey[matchKeyFor(request.trackName, request.artistName)]; - }) - .toList(growable: false); + final normalized = normalizeSpotifyId(candidate); + final normalizedKey = 'spotify_id_norm\u0000$normalized'; + if (normalized.isNotEmpty && seen.add(normalizedKey)) { + addCandidate( + requestIndex, + priority++, + 'spotify_id_norm', + normalized, + ); + } + } + addCandidate( + requestIndex, + priority++, + 'isrc_norm', + normalizeIsrc(request.isrc), + ); + addCandidate( + requestIndex, + priority, + 'match_key', + matchKeyFor(request.trackName, request.artistName), + ); + } + if (candidateRows.isEmpty) continue; + + final rows = await db.rawQuery(''' + WITH candidates(request_index, priority, lookup_kind, lookup_value) AS ( + VALUES ${candidateRows.join(', ')} + ), matches AS ( + SELECT c.request_index, c.priority, h.id, h.file_path, + h.sort_added + FROM candidates c + JOIN history h ON h.spotify_id = c.lookup_value + WHERE c.lookup_kind = 'spotify_id' + UNION ALL + SELECT c.request_index, c.priority, h.id, h.file_path, + h.sort_added + FROM candidates c + JOIN history h ON h.spotify_id_norm = c.lookup_value + WHERE c.lookup_kind = 'spotify_id_norm' + UNION ALL + SELECT c.request_index, c.priority, h.id, h.file_path, + h.sort_added + FROM candidates c + JOIN history h ON h.isrc_norm = c.lookup_value + WHERE c.lookup_kind = 'isrc_norm' + UNION ALL + SELECT c.request_index, c.priority, h.id, h.file_path, + h.sort_added + FROM candidates c + JOIN history h ON h.match_key = c.lookup_value + WHERE c.lookup_kind = 'match_key' + ) + SELECT request_index, id, file_path + FROM matches + ORDER BY request_index, priority, sort_added DESC, id DESC + ''', args); + for (final row in rows) { + final localIndex = (row['request_index'] as num).toInt(); + final resultIndex = chunkStart + localIndex; + results[resultIndex] ??= { + 'id': row['id'], + 'filePath': _normalizeIosPath(row['file_path'] as String?), + }; + } + } + return results; } Future deleteById(String id) async { diff --git a/lib/services/library_database.dart b/lib/services/library_database.dart index 7236b7e1..270ae359 100644 --- a/lib/services/library_database.dart +++ b/lib/services/library_database.dart @@ -18,10 +18,18 @@ 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 int schemaVersion = 14; static const String legacySourceId = LocalLibraryItem.legacySourceId; static const String visibleLibraryView = 'library_visible'; static const String searchFtsTable = 'library_search_fts'; + static const String lookupSummaryTable = 'library_lookup_summary'; + static const String _scanStageTable = 'library_scan_stage'; + static const String _scanStagePathKeysTable = 'library_scan_path_keys_stage'; + static const String _incrementalStageTable = 'library_incremental_stage'; + static const String _incrementalStagePathKeysTable = + 'library_incremental_path_keys_stage'; + static const String _downloadedLibraryIdsStageTable = + 'library_downloaded_ids_stage'; static const int audioMetadataScanVersion = 3; static final sqlite.SingleFlightInitializer _database = sqlite.SingleFlightInitializer(); @@ -40,6 +48,9 @@ class LibraryDatabase { onCreate: _createDB, onUpgrade: _upgradeDB, ); + // Library upserts use INSERT OR REPLACE. Recursive triggers ensure the + // implicit delete also decrements materialized lookup ref-counts. + await db.execute('PRAGMA recursive_triggers = ON'); // 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. @@ -127,6 +138,7 @@ class LibraryDatabase { await _createQueueIndexes(db); await _createPathKeyTable(db); await _createLibrarySources(db); + await _createLookupSummary(db); _searchFtsAvailable = await _createSearchFts(db); _log.i('Library database schema created with indexes'); @@ -235,6 +247,130 @@ class LibraryDatabase { ); _log.i('Added indexed lyrics availability metadata'); } + if (oldVersion < 14) { + await _createLookupSummary(db); + _log.i('Added incremental Library lookup summary'); + } + } + + Future _createLookupSummary(DatabaseExecutor db) async { + await db.execute(''' + CREATE TABLE IF NOT EXISTS $lookupSummaryTable ( + source_id TEXT NOT NULL, + kind TEXT NOT NULL, + value TEXT NOT NULL, + ref_count INTEGER NOT NULL, + PRIMARY KEY (source_id, kind, value) + ) + '''); + await db.execute( + 'CREATE INDEX IF NOT EXISTS idx_library_lookup_summary_value ' + 'ON $lookupSummaryTable(kind, value)', + ); + + for (final name in const [ + 'library_lookup_insert_isrc', + 'library_lookup_delete_isrc', + 'library_lookup_update_isrc', + 'library_lookup_insert_match', + 'library_lookup_delete_match', + 'library_lookup_update_match', + ]) { + await db.execute('DROP TRIGGER IF EXISTS $name'); + } + + await db.execute(''' + CREATE TRIGGER library_lookup_insert_isrc AFTER INSERT ON library + WHEN NEW.isrc IS NOT NULL AND NEW.isrc != '' + BEGIN + INSERT OR IGNORE INTO $lookupSummaryTable(source_id, kind, value, ref_count) + VALUES (NEW.source_id, 'isrc', NEW.isrc, 0); + UPDATE $lookupSummaryTable SET ref_count = ref_count + 1 + WHERE source_id = NEW.source_id AND kind = 'isrc' AND value = NEW.isrc; + END + '''); + await db.execute(''' + CREATE TRIGGER library_lookup_delete_isrc AFTER DELETE ON library + WHEN OLD.isrc IS NOT NULL AND OLD.isrc != '' + BEGIN + UPDATE $lookupSummaryTable SET ref_count = ref_count - 1 + WHERE source_id = OLD.source_id AND kind = 'isrc' AND value = OLD.isrc; + DELETE FROM $lookupSummaryTable + WHERE source_id = OLD.source_id AND kind = 'isrc' AND value = OLD.isrc + AND ref_count <= 0; + END + '''); + await db.execute(''' + CREATE TRIGGER library_lookup_update_isrc AFTER UPDATE OF source_id, isrc ON library + WHEN OLD.source_id IS NOT NEW.source_id OR OLD.isrc IS NOT NEW.isrc + BEGIN + UPDATE $lookupSummaryTable SET ref_count = ref_count - 1 + WHERE OLD.isrc IS NOT NULL AND OLD.isrc != '' + AND source_id = OLD.source_id AND kind = 'isrc' AND value = OLD.isrc; + DELETE FROM $lookupSummaryTable + WHERE source_id = OLD.source_id AND kind = 'isrc' AND value = OLD.isrc + AND ref_count <= 0; + INSERT OR IGNORE INTO $lookupSummaryTable(source_id, kind, value, ref_count) + SELECT NEW.source_id, 'isrc', NEW.isrc, 0 + WHERE NEW.isrc IS NOT NULL AND NEW.isrc != ''; + UPDATE $lookupSummaryTable SET ref_count = ref_count + 1 + WHERE NEW.isrc IS NOT NULL AND NEW.isrc != '' + AND source_id = NEW.source_id AND kind = 'isrc' AND value = NEW.isrc; + END + '''); + await db.execute(''' + CREATE TRIGGER library_lookup_insert_match AFTER INSERT ON library + WHEN NEW.match_key IS NOT NULL AND NEW.match_key != '' + BEGIN + INSERT OR IGNORE INTO $lookupSummaryTable(source_id, kind, value, ref_count) + VALUES (NEW.source_id, 'match', NEW.match_key, 0); + UPDATE $lookupSummaryTable SET ref_count = ref_count + 1 + WHERE source_id = NEW.source_id AND kind = 'match' AND value = NEW.match_key; + END + '''); + await db.execute(''' + CREATE TRIGGER library_lookup_delete_match AFTER DELETE ON library + WHEN OLD.match_key IS NOT NULL AND OLD.match_key != '' + BEGIN + UPDATE $lookupSummaryTable SET ref_count = ref_count - 1 + WHERE source_id = OLD.source_id AND kind = 'match' AND value = OLD.match_key; + DELETE FROM $lookupSummaryTable + WHERE source_id = OLD.source_id AND kind = 'match' AND value = OLD.match_key + AND ref_count <= 0; + END + '''); + await db.execute(''' + CREATE TRIGGER library_lookup_update_match AFTER UPDATE OF source_id, match_key ON library + WHEN OLD.source_id IS NOT NEW.source_id OR OLD.match_key IS NOT NEW.match_key + BEGIN + UPDATE $lookupSummaryTable SET ref_count = ref_count - 1 + WHERE OLD.match_key IS NOT NULL AND OLD.match_key != '' + AND source_id = OLD.source_id AND kind = 'match' AND value = OLD.match_key; + DELETE FROM $lookupSummaryTable + WHERE source_id = OLD.source_id AND kind = 'match' AND value = OLD.match_key + AND ref_count <= 0; + INSERT OR IGNORE INTO $lookupSummaryTable(source_id, kind, value, ref_count) + SELECT NEW.source_id, 'match', NEW.match_key, 0 + WHERE NEW.match_key IS NOT NULL AND NEW.match_key != ''; + UPDATE $lookupSummaryTable SET ref_count = ref_count + 1 + WHERE NEW.match_key IS NOT NULL AND NEW.match_key != '' + AND source_id = NEW.source_id AND kind = 'match' AND value = NEW.match_key; + END + '''); + + await db.delete(lookupSummaryTable); + await db.rawInsert(''' + INSERT INTO $lookupSummaryTable(source_id, kind, value, ref_count) + SELECT source_id, 'isrc', isrc, COUNT(*) + FROM library WHERE isrc IS NOT NULL AND isrc != '' + GROUP BY source_id, isrc + '''); + await db.rawInsert(''' + INSERT INTO $lookupSummaryTable(source_id, kind, value, ref_count) + SELECT source_id, 'match', match_key, COUNT(*) + FROM library WHERE match_key IS NOT NULL AND match_key != '' + GROUP BY source_id, match_key + '''); } Future _createSearchFts(DatabaseExecutor db) { @@ -607,6 +743,161 @@ class LibraryDatabase { _log.i('Batch inserted ${items.length} items'); } + /// Removes rows from one source whose normalized path keys already exist in + /// download History, without materializing either database's paths in Dart. + Future deleteDownloadedRowsForSource(String sourceId) async { + final db = await database; + await _ensureHistoryAttached(db); + const downloadedMatch = ''' + EXISTS ( + SELECT 1 + FROM library_path_keys lk + JOIN history_db.history_path_keys hk ON hk.path_key = lk.path_key + WHERE lk.item_id = library.id + ) + '''; + await db.execute('DROP TABLE IF EXISTS $_downloadedLibraryIdsStageTable'); + await db.execute( + 'CREATE TEMP TABLE $_downloadedLibraryIdsStageTable ' + '(id TEXT PRIMARY KEY)', + ); + try { + await db.rawInsert( + 'INSERT INTO $_downloadedLibraryIdsStageTable(id) ' + 'SELECT id FROM library ' + 'WHERE source_id = ? AND $downloadedMatch', + [sourceId], + ); + final countRows = await db.rawQuery( + 'SELECT COUNT(*) AS count FROM $_downloadedLibraryIdsStageTable', + ); + final count = Sqflite.firstIntValue(countRows) ?? 0; + if (count == 0) return 0; + + await db.transaction((txn) async { + await txn.rawDelete( + 'DELETE FROM library_path_keys WHERE item_id IN ' + '(SELECT id FROM $_downloadedLibraryIdsStageTable)', + ); + await txn.rawDelete( + 'DELETE FROM library WHERE id IN ' + '(SELECT id FROM $_downloadedLibraryIdsStageTable)', + ); + }); + return count; + } finally { + await db.execute('DROP TABLE IF EXISTS $_downloadedLibraryIdsStageTable'); + } + } + + /// Upserts incremental scan rows after filtering them through the attached + /// History path-key index. This keeps the hot incremental path independent + /// of total History size. + Future<({int upserted, int skipped})> upsertBatchExcludingHistory( + List> items, { + required String sourceId, + }) async { + if (items.isEmpty) return (upserted: 0, skipped: 0); + final db = await database; + await _ensureHistoryAttached(db); + await db.execute('DROP TABLE IF EXISTS $_incrementalStageTable'); + await db.execute('DROP TABLE IF EXISTS $_incrementalStagePathKeysTable'); + await db.execute( + 'CREATE TEMP TABLE $_incrementalStageTable ' + 'AS SELECT * FROM library WHERE 0', + ); + await db.execute( + 'CREATE UNIQUE INDEX idx_${_incrementalStageTable}_id ' + 'ON $_incrementalStageTable(id)', + ); + await db.execute( + 'CREATE UNIQUE INDEX idx_${_incrementalStageTable}_path ' + 'ON $_incrementalStageTable(file_path)', + ); + await db.execute(''' + CREATE TEMP TABLE $_incrementalStagePathKeysTable ( + item_id TEXT NOT NULL, + path_key TEXT NOT NULL, + PRIMARY KEY (item_id, path_key) + ) + '''); + await db.execute( + 'CREATE INDEX idx_${_incrementalStagePathKeysTable}_key ' + 'ON $_incrementalStagePathKeysTable(path_key)', + ); + + try { + final batch = db.batch(); + for (final json in items) { + final id = json['id'] as String?; + if (id == null || id.trim().isEmpty) { + throw const FormatException('Library scan row has no valid id'); + } + batch.insert( + _incrementalStageTable, + _jsonToDbRow(json, sourceId: sourceId), + conflictAlgorithm: ConflictAlgorithm.replace, + ); + sqlite.putPathKeysInBatch( + batch, + _incrementalStagePathKeysTable, + id, + json['filePath'] as String?, + ); + } + await batch.commit(noResult: true); + + const historyMatch = + ''' + EXISTS ( + SELECT 1 + FROM $_incrementalStagePathKeysTable sk + JOIN history_db.history_path_keys hk ON hk.path_key = sk.path_key + WHERE sk.item_id = s.id + ) + '''; + final skippedRows = await db.rawQuery( + 'SELECT COUNT(*) AS count FROM $_incrementalStageTable s ' + 'WHERE $historyMatch', + ); + final skipped = Sqflite.firstIntValue(skippedRows) ?? 0; + final stagedRows = await db.rawQuery( + 'SELECT COUNT(*) AS count FROM $_incrementalStageTable', + ); + final staged = Sqflite.firstIntValue(stagedRows) ?? 0; + final columns = (await db.rawQuery( + 'PRAGMA table_info(library)', + )).map((row) => row['name'] as String).toList(growable: false); + final columnList = columns.join(', '); + final selectedColumns = columns.map((column) => 's.$column').join(', '); + + await db.transaction((txn) async { + await txn.rawDelete(''' + DELETE FROM library_path_keys + WHERE item_id IN ( + SELECT s.id FROM $_incrementalStageTable s WHERE NOT $historyMatch + ) + '''); + await txn.rawInsert(''' + INSERT OR REPLACE INTO library ($columnList) + SELECT $selectedColumns FROM $_incrementalStageTable s + WHERE NOT $historyMatch + '''); + await txn.rawInsert(''' + INSERT OR IGNORE INTO library_path_keys(item_id, path_key) + SELECT sk.item_id, sk.path_key + FROM $_incrementalStagePathKeysTable sk + JOIN $_incrementalStageTable s ON s.id = sk.item_id + WHERE NOT $historyMatch + '''); + }); + return (upserted: staged - skipped, skipped: skipped); + } finally { + await db.execute('DROP TABLE IF EXISTS $_incrementalStagePathKeysTable'); + await db.execute('DROP TABLE IF EXISTS $_incrementalStageTable'); + } + } + Future replaceAll(List> items) async { final db = await database; await db.transaction((txn) async { @@ -680,9 +971,10 @@ class LibraryDatabase { return inserted; } - /// Atomically replaces only one source. Other folders, including temporarily - /// disconnected removable storage, retain their index rows. - Future replaceSourceStream( + /// Stages scan rows in bounded, independently committed batches, then swaps + /// only this source in one short transaction. Download-history exclusion is + /// an indexed SQLite anti-join, avoiding a full History path set in Dart. + Future<({int inserted, int skipped})> replaceSourceStream( String sourceId, Stream> items, { int batchSize = 300, @@ -691,25 +983,40 @@ class LibraryDatabase { throw ArgumentError.value(batchSize, 'batchSize', 'Must be positive'); } final db = await database; - var inserted = 0; - await db.transaction((txn) async { - await txn.rawDelete( - 'DELETE FROM library_path_keys WHERE item_id IN ' - '(SELECT id FROM library WHERE source_id = ?)', - [sourceId], - ); - await txn.delete( - 'library', - where: 'source_id = ?', - whereArgs: [sourceId], - ); + await _ensureHistoryAttached(db); + await db.execute('DROP TABLE IF EXISTS $_scanStageTable'); + await db.execute('DROP TABLE IF EXISTS $_scanStagePathKeysTable'); + await db.execute( + 'CREATE TEMP TABLE $_scanStageTable AS SELECT * FROM library WHERE 0', + ); + await db.execute( + 'CREATE UNIQUE INDEX idx_${_scanStageTable}_id ' + 'ON $_scanStageTable(id)', + ); + await db.execute( + 'CREATE UNIQUE INDEX idx_${_scanStageTable}_path ' + 'ON $_scanStageTable(file_path)', + ); + await db.execute(''' + CREATE TEMP TABLE $_scanStagePathKeysTable ( + item_id TEXT NOT NULL, + path_key TEXT NOT NULL, + PRIMARY KEY (item_id, path_key) + ) + '''); + await db.execute( + 'CREATE INDEX idx_${_scanStagePathKeysTable}_key ' + 'ON $_scanStagePathKeysTable(path_key)', + ); - var batch = txn.batch(); + var streamed = 0; + try { + var batch = db.batch(); var pending = 0; Future flush() async { if (pending == 0) return; await batch.commit(noResult: true); - batch = txn.batch(); + batch = db.batch(); pending = 0; } @@ -719,19 +1026,87 @@ class LibraryDatabase { throw const FormatException('Library scan row has no valid id'); } batch.insert( - 'library', + _scanStageTable, _jsonToDbRow(json, sourceId: sourceId), conflictAlgorithm: ConflictAlgorithm.replace, ); - _putPathKeysInBatch(batch, id, json['filePath'] as String?); - inserted++; + sqlite.putPathKeysInBatch( + batch, + _scanStagePathKeysTable, + id, + json['filePath'] as String?, + ); + streamed++; pending++; if (pending >= batchSize) await flush(); } await flush(); - }); - _log.i('Stream-replaced library source $sourceId with $inserted items'); - return inserted; + + const historyMatch = + ''' + EXISTS ( + SELECT 1 + FROM $_scanStagePathKeysTable sk + JOIN history_db.history_path_keys hk ON hk.path_key = sk.path_key + WHERE sk.item_id = s.id + ) + '''; + final skippedRows = await db.rawQuery( + 'SELECT COUNT(*) AS count FROM $_scanStageTable s ' + 'WHERE $historyMatch', + ); + final skipped = Sqflite.firstIntValue(skippedRows) ?? 0; + final stagedRows = await db.rawQuery( + 'SELECT COUNT(*) AS count FROM $_scanStageTable', + ); + final staged = Sqflite.firstIntValue(stagedRows) ?? 0; + final columns = (await db.rawQuery( + 'PRAGMA table_info(library)', + )).map((row) => row['name'] as String).toList(growable: false); + final columnList = columns.join(', '); + final selectedColumns = columns.map((column) => 's.$column').join(', '); + + await db.transaction((txn) async { + await txn.rawDelete( + 'DELETE FROM library_path_keys WHERE item_id IN ' + '(SELECT id FROM library WHERE source_id = ?)', + [sourceId], + ); + await txn.delete( + 'library', + where: 'source_id = ?', + whereArgs: [sourceId], + ); + await txn.rawDelete(''' + DELETE FROM library_path_keys + WHERE item_id IN ( + SELECT s.id FROM $_scanStageTable s WHERE NOT $historyMatch + ) + '''); + await txn.rawInsert(''' + INSERT OR REPLACE INTO library ($columnList) + SELECT $selectedColumns FROM $_scanStageTable s + WHERE NOT $historyMatch + '''); + await txn.rawInsert(''' + INSERT OR IGNORE INTO library_path_keys(item_id, path_key) + SELECT sk.item_id, sk.path_key + FROM $_scanStagePathKeysTable sk + JOIN $_scanStageTable s ON s.id = sk.item_id + WHERE NOT $historyMatch + '''); + }); + final inserted = staged - skipped; + _log.i( + 'Streamed $streamed rows, staged $staged unique rows, and swapped ' + '$inserted into Library source ' + '$sourceId ($skipped downloads excluded)', + ); + return (inserted: inserted, skipped: skipped); + } finally { + await db.execute('DROP TABLE IF EXISTS $_scanStagePathKeysTable'); + await db.execute('DROP TABLE IF EXISTS $_scanStageTable'); + } } Future> getSources() async { @@ -1323,19 +1698,22 @@ class LibraryDatabase { Future getLookupIndex() async { final db = await database; - final rows = await db.rawQuery( - 'SELECT isrc, match_key FROM $visibleLibraryView', - ); + final rows = await db.rawQuery(''' + SELECT summary.kind, summary.value + FROM $lookupSummaryTable summary + JOIN library_sources source ON source.id = summary.source_id + WHERE source.enabled = 1 AND source.available = 1 + GROUP BY summary.kind, summary.value + '''); final isrcs = {}; final matchKeys = {}; for (final row in rows) { - final isrc = row['isrc'] as String?; - if (isrc != null && isrc.isNotEmpty) { - isrcs.add(isrc); - } - final matchKey = row['match_key'] as String?; - if (matchKey != null && matchKey.isNotEmpty) { - matchKeys.add(matchKey); + final value = row['value'] as String?; + if (value == null || value.isEmpty) continue; + if (row['kind'] == 'isrc') { + isrcs.add(value); + } else if (row['kind'] == 'match') { + matchKeys.add(value); } } return LocalLibraryLookupIndex(