mirror of
https://github.com/zarzet/SpotiFLAC-Mobile.git
synced 2026-09-02 16:20:57 +02:00
perf: optimize library scans and extension runtime
This commit is contained in:
zarzet
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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:
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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<LocalLibraryState> {
|
||||
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<LocalLibraryState> {
|
||||
return reconnected;
|
||||
}
|
||||
|
||||
bool _isDownloadedPath(String? filePath, Set<String> 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<String> downloadedPathKeys,
|
||||
}) async {
|
||||
if (_scanCancelRequested) return null;
|
||||
final scanFile = isSaf
|
||||
@@ -549,7 +531,6 @@ class LocalLibraryNotifier extends Notifier<LocalLibraryState> {
|
||||
folderPath,
|
||||
isCancelled: () => _scanCancelRequested,
|
||||
);
|
||||
var skipped = 0;
|
||||
try {
|
||||
if (_scanCancelRequested) return null;
|
||||
state = state.copyWith(
|
||||
@@ -557,18 +538,13 @@ class LocalLibraryNotifier extends Notifier<LocalLibraryState> {
|
||||
scanProgress: state.scanProgress >= 99 ? state.scanProgress : 99,
|
||||
scanCurrentFile: null,
|
||||
);
|
||||
Stream<Map<String, dynamic>> filteredRows() async* {
|
||||
Stream<Map<String, dynamic>> 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<LocalLibraryState> {
|
||||
}
|
||||
}
|
||||
|
||||
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<LocalLibraryState> {
|
||||
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 = <String>{
|
||||
...downloadedPaths,
|
||||
...inMemoryHistoryPaths,
|
||||
};
|
||||
final downloadedPathKeys = <String>{};
|
||||
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<LocalLibraryState> {
|
||||
sourceId: activeSourceId,
|
||||
folderPath: effectiveFolderPath,
|
||||
isSaf: isSaf,
|
||||
downloadedPathKeys: downloadedPathKeys,
|
||||
);
|
||||
if (scanResult == null || _scanCancelRequested) {
|
||||
state = state.copyWith(
|
||||
@@ -868,39 +824,31 @@ class LocalLibraryNotifier extends Notifier<LocalLibraryState> {
|
||||
'$skippedCount skipped, ${deletedPaths.length} deleted, $totalFiles total',
|
||||
);
|
||||
|
||||
final existingPaths = existingFiles.keys.toList(growable: false);
|
||||
final existingDownloadedPaths = <String>[];
|
||||
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 = <LocalLibraryItem>[];
|
||||
int skippedDownloads = existingDownloadedPaths.length;
|
||||
var skippedDownloads = removedDownloaded;
|
||||
if (scannedList.isNotEmpty) {
|
||||
for (final json in scannedList) {
|
||||
final map = json as Map<String, dynamic>;
|
||||
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(
|
||||
|
||||
@@ -554,38 +554,47 @@ class _DownloadedAlbumScreenState extends ConsumerState<DownloadedAlbumScreen>
|
||||
}
|
||||
|
||||
final discNumbers = _getSortedDiscNumbers(tracks);
|
||||
final List<Widget> children = [];
|
||||
final navigationIndexById = <String, int>{
|
||||
for (var index = 0; index < tracks.length; index++)
|
||||
tracks[index].id: index,
|
||||
};
|
||||
final slivers = <Widget>[];
|
||||
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),
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -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<String> _preparedNativeDashManifestPaths = <String>{};
|
||||
static final Map<String, _AudioProbeCacheEntry> _audioProbeCache =
|
||||
<String, _AudioProbeCacheEntry>{};
|
||||
|
||||
static String _buildOutputPath(String inputPath, String extension) {
|
||||
final normalizedExt = extension.startsWith('.') ? extension : '.$extension';
|
||||
@@ -407,21 +425,71 @@ class FFmpegService {
|
||||
}
|
||||
|
||||
static Future<String?> probePrimaryAudioCodec(String filePath) async {
|
||||
return (await _probePrimaryAudioProperties(filePath)).codec;
|
||||
}
|
||||
|
||||
static Future<String?> _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 <String, dynamic>{};
|
||||
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<int?> 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 <String, dynamic>{};
|
||||
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<int?> 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 <String, dynamic>{};
|
||||
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 &&
|
||||
|
||||
@@ -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<List<Map<String, dynamic>?>> findExistingTracks(
|
||||
List<HistoryLookupRequest> requests,
|
||||
) async {
|
||||
if (requests.isEmpty) return const [];
|
||||
final db = await database;
|
||||
final bySpotify = <String, Map<String, dynamic>>{};
|
||||
final bySpotifyNorm = <String, Map<String, dynamic>>{};
|
||||
final byIsrcNorm = <String, Map<String, dynamic>>{};
|
||||
final byMatchKey = <String, Map<String, dynamic>>{};
|
||||
final results = List<Map<String, dynamic>?>.filled(requests.length, null);
|
||||
const requestChunkSize = 80;
|
||||
|
||||
Future<void> loadColumn(
|
||||
String column,
|
||||
Iterable<String> rawValues,
|
||||
Map<String, Map<String, dynamic>> 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 = <String>[];
|
||||
final args = <Object?>[];
|
||||
|
||||
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 = <String>{};
|
||||
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<void> deleteById(String id) async {
|
||||
|
||||
@@ -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> _database =
|
||||
sqlite.SingleFlightInitializer<Database>();
|
||||
@@ -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<void> _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<bool> _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<int> 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<Map<String, dynamic>> 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<void> replaceAll(List<Map<String, dynamic>> 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<int> 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<Map<String, dynamic>> 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<void> 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<List<LocalLibrarySource>> getSources() async {
|
||||
@@ -1323,19 +1698,22 @@ class LibraryDatabase {
|
||||
|
||||
Future<LocalLibraryLookupIndex> 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 = <String>{};
|
||||
final matchKeys = <String>{};
|
||||
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(
|
||||
|
||||
Reference in New Issue
Block a user