package credits import ( "context" "errors" "io" "log/slog" "sync" "sync/atomic" "testing" "time" ) // The service, and the claim it exists to keep: in the settled case a candidate costs one // indexed read and touches no media at all. // // The fake detector counts every call, so "no file access" is an assertion rather than a // hope — which is the only way to test a property whose failure is invisible. type fakeRepo struct { mu sync.Mutex markers map[string]Marker season []Marker writes int32 reads int32 } func newFakeRepo() *fakeRepo { return &fakeRepo{markers: map[string]Marker{}} } func (r *fakeRepo) key(itemID, fingerprint string) string { return itemID + "|" + fingerprint } func (r *fakeRepo) GetMarker(_ context.Context, itemID, fingerprint string) (Marker, bool, error) { atomic.AddInt32(&r.reads, 1) r.mu.Lock() defer r.mu.Unlock() marker, found := r.markers[r.key(itemID, fingerprint)] return marker, found, nil } func (r *fakeRepo) SaveMarker(_ context.Context, marker Marker) error { atomic.AddInt32(&r.writes, 1) r.mu.Lock() defer r.mu.Unlock() r.markers[r.key(marker.ItemID, marker.MediaFingerprint)] = marker return nil } func (r *fakeRepo) SeasonMarkers(context.Context, string, int, int) ([]Marker, error) { return r.season, nil } type fakeDetector struct { calls int32 result Detection err error } func (d *fakeDetector) Detect(ctx context.Context, _ MediaInfo) (Detection, error) { atomic.AddInt32(&d.calls, 1) if d.err != nil { return Detection{}, d.err } if ctx.Err() != nil { return Detection{}, ctx.Err() } return d.result, nil } type fakeResolver struct { mu sync.Mutex media map[string]ResolvedMedia } func (r *fakeResolver) Resolve(_ context.Context, itemID string) (ResolvedMedia, error) { r.mu.Lock() defer r.mu.Unlock() media, found := r.media[itemID] if !found { return ResolvedMedia{}, errors.New("unknown item") } return media, nil } type fakeBehaviour struct{ stops []StopEvent } func (b fakeBehaviour) Stops(context.Context, string) ([]StopEvent, error) { return b.stops, nil } type fixedCandidates []Candidate func (f fixedCandidates) Candidates(context.Context) ([]Candidate, error) { return f, nil } type fakeHistory struct { recent map[string]time.Time attempts []ScanAttempt } func (h *fakeHistory) SaveScanAttempt(_ context.Context, attempt ScanAttempt) error { h.attempts = append(h.attempts, attempt) return nil } func (h *fakeHistory) RecentScanTimes( _ context.Context, _ []string, _ time.Time, ) (map[string]time.Time, error) { return h.recent, nil } func quietLog() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelError})) } func testMedia() ResolvedMedia { return ResolvedMedia{ URL: "http://emby/Videos/bb-s06e08/stream", SeriesID: "bb", Season: 6, Episode: 8, RuntimeMs: bbRuntime, Version: MediaVersion{ ItemID: "bb-s06e08", RuntimeMs: bbRuntime, SizeBytes: 1_400_000_000, ETag: "abc123", }, } } func newTestService(repo *fakeRepo, detector Detector, media ResolvedMedia) *Service { return New(Deps{ Repository: repo, Resolver: &fakeResolver{media: map[string]ResolvedMedia{media.Version.ItemID: media}}, Detector: detector, Log: quietLog(), Config: DefaultConfig(), }) } // The steady state the whole subsystem is optimised for. func TestCachedMarkerPreventsAnyMediaAccess(t *testing.T) { repo := newFakeRepo() media := testMedia() repo.markers[repo.key(media.Version.ItemID, Fingerprint(media.Version))] = Marker{ ItemID: media.Version.ItemID, CreditsStartMs: 2_450_000, Confidence: 0.93, DetectionMethod: MethodCombined, } detector := &fakeDetector{result: visual(2_450_000, 0.9)} service := newTestService(repo, detector, media) detection, stored, err := service.Process(context.Background(), media.Version.ItemID) if err != nil { t.Fatal(err) } if detector.calls != 0 { t.Fatalf("the detector ran %d times for an episode that was already decided", detector.calls) } if stored { t.Fatal("a cached marker caused a database write") } if atomic.LoadInt32(&repo.writes) != 0 { t.Fatalf("%d writes on the cached path; the promise is zero", repo.writes) } if detection.StartMs != 2_450_000 { t.Fatalf("StartMs = %d, want the stored marker", detection.StartMs) } } // A successful scan is one write, and exactly one. func TestSuccessfulScanWritesOnce(t *testing.T) { repo := newFakeRepo() media := testMedia() detector := &fakeDetector{result: visual(2_450_000, 0.85)} service := newTestService(repo, detector, media) if _, stored, err := service.Process(context.Background(), media.Version.ItemID); err != nil || !stored { t.Fatalf("first scan stored = %v, err = %v", stored, err) } if repo.writes != 1 { t.Fatalf("%d writes for one scan, want 1", repo.writes) } // And the second time the episode comes round, nothing at all. if _, stored, err := service.Process(context.Background(), media.Version.ItemID); err != nil || stored { t.Fatalf("second pass stored = %v, err = %v; the answer was already known", stored, err) } if repo.writes != 1 { t.Fatalf("%d writes after re-processing, want 1", repo.writes) } if detector.calls != 1 { t.Fatalf("the detector ran %d times; the second pass should never reach it", detector.calls) } } // The reason the marker is keyed on a fingerprint rather than an item id. func TestReplacedFileInvalidatesTheMarker(t *testing.T) { repo := newFakeRepo() media := testMedia() detector := &fakeDetector{result: visual(2_450_000, 0.85)} service := newTestService(repo, detector, media) if _, _, err := service.Process(context.Background(), media.Version.ItemID); err != nil { t.Fatal(err) } // Sonarr swaps the file: same episode, same item id, different media. replaced := media replaced.Version.SizeBytes = 2_900_000_000 replaced.Version.ETag = "def456" service = newTestService(repo, detector, replaced) if _, stored, err := service.Process(context.Background(), replaced.Version.ItemID); err != nil || !stored { t.Fatalf("a replaced file was not rescanned: stored = %v, err = %v", stored, err) } if detector.calls != 2 { t.Fatalf("the detector ran %d times; the replacement should have been scanned", detector.calls) } } // A marker that could outlive the file it describes must not be stored. func TestWeakFingerprintWithholdsTheMarker(t *testing.T) { repo := newFakeRepo() media := testMedia() media.Version = MediaVersion{ItemID: media.Version.ItemID, RuntimeMs: bbRuntime} service := newTestService(repo, &fakeDetector{result: visual(2_450_000, 0.85)}, media) if _, stored, err := service.Process(context.Background(), media.Version.ItemID); err != nil || stored { t.Fatalf("stored = %v, err = %v; a runtime-only fingerprint cannot be invalidated", stored, err) } if repo.writes != 0 { t.Fatalf("%d writes for a weak fingerprint, want 0", repo.writes) } } // Emby's own answer wins, and costs nothing. func TestEmbyChapterMarkerSkipsTheScanEntirely(t *testing.T) { repo := newFakeRepo() media := testMedia() media.EmbyCreditsMs = 2_460_000 detector := &fakeDetector{result: visual(2_450_000, 0.9)} service := newTestService(repo, detector, media) detection, stored, err := service.Process(context.Background(), media.Version.ItemID) if err != nil { t.Fatal(err) } if detector.calls != 0 || stored || repo.writes != 0 { t.Fatalf("Emby had already answered but the scan ran anyway "+ "(calls=%d stored=%v writes=%d)", detector.calls, stored, repo.writes) } if detection.StartMs != 2_460_000 || detection.Method != MethodEmby { t.Fatalf("detection = %+v, want Emby's own marker", detection) } } // A weak reading is a scan that deliberately produces no database activity. func TestSubThresholdDetectionIsNotStored(t *testing.T) { repo := newFakeRepo() media := testMedia() service := newTestService(repo, &fakeDetector{result: visual(2_450_000, 0.4)}, media) if _, stored, err := service.Process(context.Background(), media.Version.ItemID); err != nil || stored { t.Fatalf("stored = %v, err = %v; a doubtful reading must not be written", stored, err) } if repo.writes != 0 { t.Fatalf("%d writes for a rejected detection, want 0", repo.writes) } } // No decoder is a deliberate deployment, not a fault: behaviour alone still answers. func TestMissingFFmpegFallsBackToBehaviour(t *testing.T) { repo := newFakeRepo() media := testMedia() service := New(Deps{ Repository: repo, Resolver: &fakeResolver{media: map[string]ResolvedMedia{media.Version.ItemID: media}}, Detector: &fakeDetector{err: ErrNoFFmpeg}, Behaviour: fakeBehaviour{stops: []StopEvent{ {UserKey: "paul", PositionMs: 2_450_000, RuntimeMs: bbRuntime, NextEpisode: true}, {UserKey: "david", PositionMs: 2_453_000, RuntimeMs: bbRuntime, NextEpisode: true}, {UserKey: "matt", PositionMs: 2_451_000, RuntimeMs: bbRuntime, NextEpisode: true}, }}, Log: quietLog(), Config: DefaultConfig(), }) detection, stored, err := service.Process(context.Background(), media.Version.ItemID) if err != nil { t.Fatalf("a missing decoder was reported as an error: %v", err) } if !stored || detection.Method != MethodBehaviour { t.Fatalf("stored = %v, method = %q; behaviour alone should have answered", stored, detection.Method) } } // Four signals naming one episode must collapse into one scan. func TestConcurrentRequestsCollapseIntoOneScan(t *testing.T) { repo := newFakeRepo() media := testMedia() slow := &slowDetector{result: visual(2_450_000, 0.85)} service := newTestService(repo, slow, media) var group sync.WaitGroup for index := 0; index < 6; index++ { group.Add(1) go func() { defer group.Done() _, _, _ = service.Process(context.Background(), media.Version.ItemID) }() } group.Wait() if calls := atomic.LoadInt32(&slow.calls); calls != 1 { t.Fatalf("six simultaneous requests produced %d scans, want 1", calls) } if repo.writes != 1 { t.Fatalf("%d writes, want 1", repo.writes) } } type slowDetector struct { calls int32 result Detection } func (d *slowDetector) Detect(ctx context.Context, _ MediaInfo) (Detection, error) { atomic.AddInt32(&d.calls, 1) select { case <-ctx.Done(): return Detection{}, ctx.Err() case <-time.After(40 * time.Millisecond): } return d.result, nil } func TestCancellationStopsAScan(t *testing.T) { repo := newFakeRepo() media := testMedia() service := newTestService(repo, &slowDetector{result: visual(2_450_000, 0.9)}, media) ctx, cancel := context.WithCancel(context.Background()) cancel() if _, stored, err := service.Process(ctx, media.Version.ItemID); err == nil || stored { t.Fatalf("a cancelled scan returned stored = %v, err = %v", stored, err) } if repo.writes != 0 { t.Fatalf("%d writes after cancellation, want 0", repo.writes) } } // The delay is what stops a curious button press becoming disk activity. func TestAbandonedPlaybackIsNeverQueued(t *testing.T) { media := testMedia() service := newTestService(newFakeRepo(), &fakeDetector{}, media) service.liveDelay = 50 * time.Millisecond service.NotePlayback(context.Background(), media.Version.ItemID) service.AbandonPlayback(media.Version.ItemID) time.Sleep(120 * time.Millisecond) if depth := service.QueueDepth(); depth != 0 { t.Fatalf("queue depth = %d; an abandoned playback was queued anyway", depth) } } func TestSustainedPlaybackIsQueuedAtLivePriority(t *testing.T) { media := testMedia() service := newTestService(newFakeRepo(), &fakeDetector{}, media) service.liveDelay = 20 * time.Millisecond service.NotePlayback(context.Background(), media.Version.ItemID) deadline := time.Now().Add(time.Second) for service.QueueDepth() == 0 && time.Now().Before(deadline) { time.Sleep(5 * time.Millisecond) } pending := service.Pending() if len(pending) != 1 { t.Fatalf("queue holds %d candidates, want 1", len(pending)) } if pending[0].Priority != PriorityLive || pending[0].Reason != ReasonLivePlayback { t.Fatalf("candidate = %+v, want live playback at priority %d", pending[0], PriorityLive) } } // An inconclusive scan used to be forgotten, so the ten-minute refresh selected the same // episode for ever. History now makes that episode ineligible until its cooldown expires. func TestRefreshDefersRecentlyScannedCandidate(t *testing.T) { media := testMedia() history := &fakeHistory{recent: map[string]time.Time{media.Version.ItemID: time.Now()}} service := New(Deps{ Repository: newFakeRepo(), History: history, Resolver: &fakeResolver{media: map[string]ResolvedMedia{media.Version.ItemID: media}}, Source: fixedCandidates{{ItemID: media.Version.ItemID, Priority: PriorityNext}}, Log: quietLog(), Config: DefaultConfig(), }) detail, err := service.Refresh(context.Background()) if err != nil { t.Fatal(err) } if service.QueueDepth() != 0 { t.Fatal("a recently scanned episode returned to the queue") } if detail != "1 candidate cooling down, 0 already known" { t.Fatalf("refresh detail = %q", detail) } } func TestScanAttemptHistoryDescribesNoMatch(t *testing.T) { history := &fakeHistory{} service := New(Deps{ Repository: newFakeRepo(), History: history, Resolver: &fakeResolver{media: map[string]ResolvedMedia{}}, Log: quietLog(), Config: DefaultConfig(), }) candidate := Candidate{ItemID: "episode-7", Reason: ReasonNext, Priority: PriorityNext} service.recordAttempt(context.Background(), candidate, Detection{FramesSampled: 42}, false, nil, time.Now().Add(-time.Second)) if len(history.attempts) != 1 || history.attempts[0].Outcome != "no_match" || history.attempts[0].Frames != 42 { t.Fatalf("history = %+v", history.attempts) } }