memby/server/internal/credits/service_test.go

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 }

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)
	}
}