diff --git a/go_backend/cue_duplicate_supplement_test.go b/go_backend/cue_duplicate_supplement_test.go index 34cc19cf..a49ff8db 100644 --- a/go_backend/cue_duplicate_supplement_test.go +++ b/go_backend/cue_duplicate_supplement_test.go @@ -117,6 +117,101 @@ func TestCueParserEndToEnd(t *testing.T) { } } +func writeTestFlacWithISRC(t *testing.T, path, isrc string) { + t.Helper() + le32 := func(v int) []byte { + return []byte{byte(v), byte(v >> 8), byte(v >> 16), byte(v >> 24)} + } + vendor := "test-vendor" + comments := [][]byte{ + []byte("TITLE=Song"), + []byte("ISRC=" + isrc), + } + var vorbis []byte + vorbis = append(vorbis, le32(len(vendor))...) + vorbis = append(vorbis, vendor...) + vorbis = append(vorbis, le32(len(comments))...) + for _, comment := range comments { + vorbis = append(vorbis, le32(len(comment))...) + vorbis = append(vorbis, comment...) + } + + blockHeader := func(last bool, blockType byte, length int) []byte { + first := blockType + if last { + first |= 0x80 + } + return []byte{first, byte(length >> 16), byte(length >> 8), byte(length)} + } + + var data []byte + data = append(data, "fLaC"...) + data = append(data, blockHeader(false, 0, 34)...) // STREAMINFO + data = append(data, make([]byte, 34)...) + picture := make([]byte, 4096) // stands in for embedded cover art + data = append(data, blockHeader(false, 6, len(picture))...) + data = append(data, picture...) + data = append(data, blockHeader(true, 4, len(vorbis))...) + data = append(data, vorbis...) + + if err := os.WriteFile(path, data, 0600); err != nil { + t.Fatal(err) + } +} + +func TestISRCIndexIncrementalRebuild(t *testing.T) { + dir := t.TempDir() + trackA := filepath.Join(dir, "a.flac") + trackB := filepath.Join(dir, "b.flac") + writeTestFlacWithISRC(t, trackA, "USAA00000001") + writeTestFlacWithISRC(t, trackB, "USBB00000002") + defer InvalidateISRCCache(dir) + + if got := readFlacISRC(trackA); got != "USAA00000001" { + t.Fatalf("readFlacISRC = %q", got) + } + if got := readFlacISRC(filepath.Join(dir, "missing.flac")); got != "" { + t.Fatalf("expected empty ISRC for missing file, got %q", got) + } + + idx := buildISRCIndex(dir) + if path, ok := idx.lookup("usaa00000001"); !ok || path != trackA { + t.Fatalf("lookup A = %q/%v", path, ok) + } + if path, ok := idx.lookup("USBB00000002"); !ok || path != trackB { + t.Fatalf("lookup B = %q/%v", path, ok) + } + + // Change one file's tag (and mtime); an incremental rebuild must pick up + // the change while adopting the untouched file from the cache. + writeTestFlacWithISRC(t, trackB, "USBB00000099") + future := time.Now().Add(2 * time.Second) + if err := os.Chtimes(trackB, future, future); err != nil { + t.Fatal(err) + } + rebuilt := buildISRCIndex(dir) + if _, ok := rebuilt.lookup("USBB00000002"); ok { + t.Fatal("expected stale ISRC to disappear after rebuild") + } + if path, ok := rebuilt.lookup("USBB00000099"); !ok || path != trackB { + t.Fatalf("rebuilt lookup = %q/%v", path, ok) + } + if path, ok := rebuilt.lookup("USAA00000001"); !ok || path != trackA { + t.Fatalf("cached entry lost on rebuild: %q/%v", path, ok) + } + + // Add() keeps the index fresh so the TTL never forces a rebuild while + // downloads are actively maintaining it. + rebuilt.buildTime.Store(time.Now().Add(-isrcIndexTTL - time.Minute).UnixNano()) + if rebuilt.isFresh() { + t.Fatal("expected stale index") + } + rebuilt.Add("USCC00000003", trackA) + if !rebuilt.isFresh() { + t.Fatal("expected Add to refresh the index timestamp") + } +} + func TestDuplicateIndexAndParallelExistence(t *testing.T) { dir := t.TempDir() filePath := filepath.Join(dir, "song.flac") @@ -124,7 +219,8 @@ func TestDuplicateIndexAndParallelExistence(t *testing.T) { t.Fatal(err) } - idx := &ISRCIndex{index: map[string]string{}, outputDir: dir, buildTime: time.Now()} + idx := &ISRCIndex{index: map[string]string{}, outputDir: dir} + idx.buildTime.Store(time.Now().UnixNano()) idx.Add("usrc17607839", filePath) if got, ok := idx.lookup("USRC17607839"); !ok || got != filePath { t.Fatalf("lookup = %q/%v", got, ok) diff --git a/go_backend/duplicate.go b/go_backend/duplicate.go index 35bb2953..10d6a9c5 100644 --- a/go_backend/duplicate.go +++ b/go_backend/duplicate.go @@ -1,19 +1,31 @@ package gobackend import ( + "encoding/binary" "encoding/json" "fmt" + "io" "os" "path/filepath" "strings" "sync" + "sync/atomic" "time" ) +// isrcFileEntry caches the parse result for one file so index rebuilds only +// re-read files whose size or mtime changed. +type isrcFileEntry struct { + size int64 + modTime int64 // UnixNano + isrc string // uppercase; empty when the file carries no ISRC tag +} + type ISRCIndex struct { - index map[string]string // ISRC (uppercase) -> file path + index map[string]string // ISRC (uppercase) -> file path + files map[string]isrcFileEntry // file path -> cached parse result outputDir string - buildTime time.Time + buildTime atomic.Int64 // UnixNano of the last build or write mu sync.RWMutex } @@ -22,14 +34,20 @@ var ( isrcIndexCacheMu sync.RWMutex isrcBuildingMu sync.Map // Per-directory build lock to prevent concurrent builds isrcIndexTTL = 5 * time.Minute + + isrcIndexBuildWorkers = 4 ) +func (idx *ISRCIndex) isFresh() bool { + return time.Since(time.Unix(0, idx.buildTime.Load())) < isrcIndexTTL +} + func GetISRCIndex(outputDir string) *ISRCIndex { isrcIndexCacheMu.RLock() idx, exists := isrcIndexCache[outputDir] isrcIndexCacheMu.RUnlock() - if exists && time.Since(idx.buildTime) < isrcIndexTTL { + if exists && idx.isFresh() { return idx } @@ -42,7 +60,7 @@ func GetISRCIndex(outputDir string) *ISRCIndex { idx, exists = isrcIndexCache[outputDir] isrcIndexCacheMu.RUnlock() - if exists && time.Since(idx.buildTime) < isrcIndexTTL { + if exists && idx.isFresh() { return idx } @@ -52,16 +70,37 @@ func GetISRCIndex(outputDir string) *ISRCIndex { func buildISRCIndex(outputDir string) *ISRCIndex { idx := &ISRCIndex{ index: make(map[string]string), + files: make(map[string]isrcFileEntry), outputDir: outputDir, - buildTime: time.Now(), } + idx.buildTime.Store(time.Now().UnixNano()) if outputDir == "" { return idx } + // Reuse the previous build's per-file cache: files whose size and mtime + // are unchanged are adopted without touching their content, so a rebuild + // is normally just a stat walk. + prevFiles := map[string]isrcFileEntry{} + isrcIndexCacheMu.RLock() + if prev, ok := isrcIndexCache[outputDir]; ok { + prev.mu.RLock() + for path, entry := range prev.files { + prevFiles[path] = entry + } + prev.mu.RUnlock() + } + isrcIndexCacheMu.RUnlock() + startTime := time.Now() - fileCount := 0 + type parseTask struct { + path string + size int64 + modTime int64 + } + var toParse []parseTask + reused := 0 filepath.Walk(outputDir, func(path string, info os.FileInfo, err error) error { if err != nil || info.IsDir() { @@ -73,18 +112,56 @@ func buildISRCIndex(outputDir string) *ISRCIndex { return nil } - metadata, err := ReadMetadata(path) - if err != nil || metadata.ISRC == "" { + size := info.Size() + modTime := info.ModTime().UnixNano() + if entry, ok := prevFiles[path]; ok && entry.size == size && entry.modTime == modTime { + idx.files[path] = entry + if entry.isrc != "" { + idx.index[entry.isrc] = path + } + reused++ return nil } - - idx.index[strings.ToUpper(metadata.ISRC)] = path - fileCount++ + toParse = append(toParse, parseTask{path: path, size: size, modTime: modTime}) return nil }) - fmt.Printf("[ISRCIndex] Built index for %s: %d files in %v\n", - outputDir, fileCount, time.Since(startTime).Round(time.Millisecond)) + if len(toParse) > 0 { + // New/changed files: read only their Vorbis comment block, in + // parallel. Embedded cover art (megabytes per file) is never loaded. + isrcs := make([]string, len(toParse)) + workerCount := isrcIndexBuildWorkers + if len(toParse) < workerCount { + workerCount = len(toParse) + } + tasks := make(chan int) + var wg sync.WaitGroup + for w := 0; w < workerCount; w++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := range tasks { + isrcs[i] = strings.ToUpper(readFlacISRC(toParse[i].path)) + } + }() + } + for i := range toParse { + tasks <- i + } + close(tasks) + wg.Wait() + + for i, task := range toParse { + entry := isrcFileEntry{size: task.size, modTime: task.modTime, isrc: isrcs[i]} + idx.files[task.path] = entry + if entry.isrc != "" { + idx.index[entry.isrc] = task.path + } + } + } + + fmt.Printf("[ISRCIndex] Built index for %s: %d files (%d parsed, %d cached) in %v\n", + outputDir, len(idx.files), len(toParse), reused, time.Since(startTime).Round(time.Millisecond)) isrcIndexCacheMu.Lock() isrcIndexCache[outputDir] = idx @@ -93,6 +170,78 @@ func buildISRCIndex(outputDir string) *ISRCIndex { return idx } +// readFlacISRC extracts the ISRC Vorbis comment from a FLAC file by walking +// the metadata block headers and reading only the VORBIS_COMMENT payload; +// picture and padding blocks are seeked past, never loaded. Returns "" when +// the file is not FLAC or carries no ISRC tag. +func readFlacISRC(path string) string { + f, err := os.Open(path) + if err != nil { + return "" + } + defer f.Close() + + magic := make([]byte, 4) + if _, err := io.ReadFull(f, magic); err != nil || string(magic) != "fLaC" { + return "" + } + + header := make([]byte, 4) + for { + if _, err := io.ReadFull(f, header); err != nil { + return "" + } + last := header[0]&0x80 != 0 + blockType := header[0] & 0x7F + length := int64(header[1])<<16 | int64(header[2])<<8 | int64(header[3]) + if blockType == 4 { // VORBIS_COMMENT + if length > 16<<20 { + return "" + } + payload := make([]byte, length) + if _, err := io.ReadFull(f, payload); err != nil { + return "" + } + return vorbisCommentISRC(payload) + } + if last { + return "" + } + if _, err := f.Seek(length, io.SeekCurrent); err != nil { + return "" + } + } +} + +func vorbisCommentISRC(payload []byte) string { + if len(payload) < 8 { + return "" + } + offset := int(binary.LittleEndian.Uint32(payload[0:4])) + 4 + if offset < 4 || offset+4 > len(payload) { + return "" + } + count := int(binary.LittleEndian.Uint32(payload[offset : offset+4])) + offset += 4 + for i := 0; i < count; i++ { + if offset+4 > len(payload) { + return "" + } + commentLen := int(binary.LittleEndian.Uint32(payload[offset : offset+4])) + offset += 4 + if commentLen < 0 || offset+commentLen > len(payload) { + return "" + } + comment := payload[offset : offset+commentLen] + offset += commentLen + eq := strings.IndexByte(string(comment), '=') + if eq > 0 && strings.EqualFold(string(comment[:eq]), "ISRC") { + return strings.TrimSpace(string(comment[eq+1:])) + } + } + return "" +} + func (idx *ISRCIndex) lookup(isrc string) (string, bool) { if isrc == "" { return "", false @@ -126,10 +275,30 @@ func (idx *ISRCIndex) Add(isrc, filePath string) { return } - idx.mu.Lock() - defer idx.mu.Unlock() + upper := strings.ToUpper(isrc) + var entry *isrcFileEntry + if info, err := os.Stat(filePath); err == nil { + entry = &isrcFileEntry{ + size: info.Size(), + modTime: info.ModTime().UnixNano(), + isrc: upper, + } + } - idx.index[strings.ToUpper(isrc)] = filePath + idx.mu.Lock() + idx.index[upper] = filePath + if entry != nil { + if idx.files == nil { + idx.files = make(map[string]isrcFileEntry) + } + idx.files[filePath] = *entry + } + idx.mu.Unlock() + + // The index is write-maintained after every successful download; + // refreshing the timestamp keeps the TTL from forcing a full rebuild in + // the middle of the exact workload the index exists to serve. + idx.buildTime.Store(time.Now().UnixNano()) } func InvalidateISRCCache(outputDir string) {