tapir/cmd/tapir/scheduler_test.go

package main

import (
	"context"
	"errors"
	"io"
	"log/slog"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"git.d-ma.be/mathias/tapir/internal/adapters/store"
	"git.d-ma.be/mathias/tapir/internal/runner"
)

func quietLog() *slog.Logger {
	return slog.New(slog.NewTextHandler(io.Discard, nil))
}

// fakeLister returns a fixed user set (or an error) for the scheduler under test.
type fakeLister struct {
	users  []store.UserIdentity
	err    error
	noConn map[string]bool // users that have NOT connected a video source
}

func (f fakeLister) ListAllUsers(context.Context) ([]store.UserIdentity, error) {
	return f.users, f.err
}

// ConnectionsForUser reports a single youtube connection for every user except
// those in noConn, which return zero — the connection-less case the scheduler
// must skip instead of running (and failing to resolve a token for).
func (f fakeLister) ConnectionsForUser(_ context.Context, userID string) ([]store.Connection, error) {
	if f.noConn[userID] {
		return nil, nil
	}
	return []store.Connection{{Provider: "youtube"}}, nil
}

// countingRunUser records how many passes each user got, optionally failing for
// specific users, under a mutex so it is safe across the scheduler goroutine.
type countingRunUser struct {
	mu      sync.Mutex
	calls   map[string]int
	order   []string // userIDs in the order they were run, across all passes
	failFor map[string]bool
}

func newCountingRunUser(failFor ...string) *countingRunUser {
	c := &countingRunUser{calls: map[string]int{}, failFor: map[string]bool{}}
	for _, u := range failFor {
		c.failFor[u] = true
	}
	return c
}

func (c *countingRunUser) run(_ context.Context, userID string) (runner.Stats, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.calls[userID]++
	c.order = append(c.order, userID)
	if c.failFor[userID] {
		return runner.Stats{Errors: 1}, errors.New("boom")
	}
	return runner.Stats{Summarized: 1}, nil
}

func (c *countingRunUser) runOrder() []string {
	c.mu.Lock()
	defer c.mu.Unlock()
	return append([]string(nil), c.order...)
}

func (c *countingRunUser) count(userID string) int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.calls[userID]
}

func (c *countingRunUser) total() int {
	c.mu.Lock()
	defer c.mu.Unlock()
	n := 0
	for _, v := range c.calls {
		n += v
	}
	return n
}

func usersN(ids ...string) []store.UserIdentity {
	out := make([]store.UserIdentity, len(ids))
	for i, id := range ids {
		out[i] = store.UserIdentity{UserID: id, DexSubject: "dex|" + id}
	}
	return out
}

func TestDiscoveryPassRunsEveryUserOnce(t *testing.T) {
	lister := fakeLister{users: usersN("a", "b", "c")}
	rc := newCountingRunUser()

	stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())

	require.Equal(t, 1, rc.count("a"))
	require.Equal(t, 1, rc.count("b"))
	require.Equal(t, 1, rc.count("c"))
	require.Equal(t, 3, stats.Summarized, "stats are summed across users")
}

// Caption fetches share one per-IP budget; a fixed user order starves whoever is
// last. Each pass must rotate which user leads so the lead slot is shared.
func TestDiscoveryPassRotatesLeadUser(t *testing.T) {
	lister := fakeLister{users: usersN("a", "b", "c")}
	rc := newCountingRunUser()

	runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
	runDiscoveryPass(context.Background(), 1, lister, rc.run, quietLog())
	runDiscoveryPass(context.Background(), 2, lister, rc.run, quietLog())

	require.Equal(t, []string{"a", "b", "c", "b", "c", "a", "c", "a", "b"}, rc.runOrder(),
		"each pass left-rotates the user order so every user leads in turn")
	// Fairness: over a full rotation cycle every user ran the same number of times.
	require.Equal(t, 3, rc.count("a"))
	require.Equal(t, 3, rc.count("b"))
	require.Equal(t, 3, rc.count("c"))
}

// A connectionless orphan must not consume a rotation slot: rotation is over the
// connected users only, so two real users alternate the lead 50/50 even with a
// dead identity listed between them.
func TestDiscoveryPassRotationIgnoresConnectionlessUsers(t *testing.T) {
	lister := fakeLister{users: usersN("a", "orphan", "c"), noConn: map[string]bool{"orphan": true}}
	rc := newCountingRunUser()

	runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
	runDiscoveryPass(context.Background(), 1, lister, rc.run, quietLog())

	require.Equal(t, []string{"a", "c", "c", "a"}, rc.runOrder(),
		"only connected users rotate; the orphan never runs and never holds a slot")
	require.Equal(t, 0, rc.count("orphan"))
}

func TestDiscoveryPassSkipsUsersWithoutConnections(t *testing.T) {
	// b never connected a video source (e.g. a stale Dex-era orphan identity).
	// It must be skipped silently — not run and logged as a token error every pass.
	lister := fakeLister{users: usersN("a", "b", "c"), noConn: map[string]bool{"b": true}}
	rc := newCountingRunUser()

	stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())

	require.Equal(t, 1, rc.count("a"))
	require.Equal(t, 0, rc.count("b"), "a user with no connection must be skipped, not run")
	require.Equal(t, 1, rc.count("c"))
	require.Equal(t, 2, stats.Summarized, "only connected users contribute")
	require.Equal(t, 0, stats.Errors, "skipping is silent — no spurious error stat")
}

func TestDiscoveryPassOneUserFailureDoesNotStopOthers(t *testing.T) {
	lister := fakeLister{users: usersN("a", "b", "c")}
	rc := newCountingRunUser("b") // user b's pass errors

	stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())

	require.Equal(t, 1, rc.count("a"))
	require.Equal(t, 1, rc.count("b"))
	require.Equal(t, 1, rc.count("c"), "a failing user must not abort the rest")
	require.Equal(t, 2, stats.Summarized) // a + c
	require.Equal(t, 1, stats.Errors)     // b
}

func TestDiscoveryPassListerErrorIsContained(t *testing.T) {
	lister := fakeLister{err: errors.New("db down")}
	rc := newCountingRunUser()

	stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())

	require.Equal(t, 0, rc.total(), "no users enumerated → no passes")
	require.Equal(t, runner.Stats{}, stats)
}

func TestSchedulerIntervalZeroDisablesEntirely(t *testing.T) {
	lister := fakeLister{users: usersN("a", "b")}
	rc := newCountingRunUser()

	runScheduler(context.Background(), 0, lister, rc.run, quietLog())

	require.Equal(t, 0, rc.total(), "interval 0 must not run even a startup pass")
}

func TestSchedulerRunsStartupPassThenStopsOnCancel(t *testing.T) {
	lister := fakeLister{users: usersN("a", "b", "c")}
	rc := newCountingRunUser()

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		// A long interval so only the startup pass runs before we cancel.
		runScheduler(ctx, time.Hour, lister, rc.run, quietLog())
		close(done)
	}()

	// The startup pass is synchronous at the top of runScheduler; once all three
	// users have a pass it has completed and the loop is parked on the ticker.
	require.Eventually(t, func() bool { return rc.total() == 3 }, time.Second, 5*time.Millisecond)

	cancel()
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("scheduler did not exit after ctx cancel")
	}
	require.Equal(t, 3, rc.total(), "no extra passes fired between startup and cancel")
}