feat(discovery): trigger a discovery pass on YouTube connect

A newly connected account showed no videos until the next 2h scheduled pass —
the gap that made onboarding look broken (a second user connected, saw nothing,
read as failure). The connect callback now fires an out-of-band discovery pass
for the connecting user, so videos appear promptly.

Concurrency: scheduled and connect-triggered passes share one lock (serialize),
preserving the single-fetcher invariant (ADR-018). A trigger interleaves between
the scheduler's per-user passes rather than fetching concurrently or waiting for
a whole pass. The trigger runs on the server ctx (survives the redirect) and is
non-blocking for the request goroutine.

Scope: connect-trigger only. The optional login-refresh / "Discover now" button
from #6 are intentionally not built — an unconditional login hook risks 429
storms (per the ticket's own recommendation); defer until wanted.

TDD: TestCallbackTriggersDiscovery, TestSerializeRunsOneAtATime,
TestDiscoveryTriggerEnqueueRunsUser; new BDD scenario mapped.

Refs #6

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

cmd/tapir/discovery.go

@@ -0,0 +1,44 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+
+	"gitea.d-ma.be/mathias/tapir/internal/runner"
+)
+
+// discoveryRunner runs one user's discovery pass.
+type discoveryRunner func(ctx context.Context, userID string) (runner.Stats, error)
+
+// serialize wraps run so calls never overlap: every discovery pass — scheduled
+// or connect-triggered (#6) — acquires the same lock, preserving the
+// one-fetcher-at-a-time invariant the scheduler relies on (ADR-018, the
+// single-replica assumption). Locking is per-user, so a connect-triggered pass
+// interleaves between the scheduler's users instead of waiting for a whole pass.
+func serialize(mu *sync.Mutex, run discoveryRunner) discoveryRunner {
+	return func(ctx context.Context, userID string) (runner.Stats, error) {
+		mu.Lock()
+		defer mu.Unlock()
+		return run(ctx, userID)
+	}
+}
+
+// discoveryTrigger fires an out-of-band discovery pass for one user without
+// blocking the caller (the connect HTTP handler). The pass runs on the server's
+// long-lived ctx — not the request ctx — so it survives the post-connect
+// redirect. run is the serialized runner, so a trigger never overlaps the
+// scheduler. Satisfies web.DiscoveryTrigger.
+type discoveryTrigger struct {
+	ctx context.Context
+	run discoveryRunner
+	log *slog.Logger
+}
+
+func (t *discoveryTrigger) Enqueue(userID string) {
+	go func() {
+		if _, err := t.run(t.ctx, userID); err != nil {
+			t.log.Warn("discovery: connect-triggered pass had errors", "user", userID, "err", err)
+		}
+	}()
+}

cmd/tapir/discovery_test.go

@@ -0,0 +1,62 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"gitea.d-ma.be/mathias/tapir/internal/runner"
+)
+
+// serialize must guarantee at most one discovery pass runs at a time, so a
+// connect-triggered pass never fetches concurrently with the scheduler.
+func TestSerializeRunsOneAtATime(t *testing.T) {
+	var active, maxActive int32
+	run := func(_ context.Context, _ string) (runner.Stats, error) {
+		n := atomic.AddInt32(&active, 1)
+		for { // record the high-water mark of concurrent runs
+			m := atomic.LoadInt32(&maxActive)
+			if n <= m || atomic.CompareAndSwapInt32(&maxActive, m, n) {
+				break
+			}
+		}
+		time.Sleep(2 * time.Millisecond)
+		atomic.AddInt32(&active, -1)
+		return runner.Stats{}, nil
+	}
+
+	s := serialize(&sync.Mutex{}, run)
+	var wg sync.WaitGroup
+	for i := 0; i < 20; i++ {
+		wg.Add(1)
+		go func(i int) { defer wg.Done(); _, _ = s(context.Background(), fmt.Sprintf("u%d", i)) }(i)
+	}
+	wg.Wait()
+
+	require.Equal(t, int32(1), atomic.LoadInt32(&maxActive),
+		"serialize must run at most one pass at a time")
+}
+
+// Enqueue runs the user's pass out-of-band (non-blocking) on the trigger's ctx.
+func TestDiscoveryTriggerEnqueueRunsUser(t *testing.T) {
+	done := make(chan string, 1)
+	run := func(_ context.Context, userID string) (runner.Stats, error) {
+		done <- userID
+		return runner.Stats{}, nil
+	}
+	tr := &discoveryTrigger{ctx: context.Background(), run: run, log: quietLog()}
+
+	tr.Enqueue("u1")
+
+	select {
+	case got := <-done:
+		require.Equal(t, "u1", got)
+	case <-time.After(2 * time.Second):
+		t.Fatal("Enqueue did not run the user's pass")
+	}
+}

cmd/tapir/main.go

@@ -20,6 +20,7 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"sync"
 	"time"
 
 	"gitea.d-ma.be/mathias/tapir/internal/adapters/secrets"
@@ -238,13 +239,20 @@ func cmdServe(ctx context.Context, log *slog.Logger) error {
 	if cfg.DiscoveryInterval > 0 {
 		log.Info("scheduled discovery enabled", "interval", cfg.DiscoveryInterval, "fetch_rate", cfg.FetchRate)
 		log.Warn("scheduled discovery assumes a SINGLE replica — running serve at >1 replica double-runs discovery (ADR-018)")
-		runUser := func(ctx context.Context, userID string) (runner.Stats, error) {
+		rawRunUser := func(ctx context.Context, userID string) (runner.Stats, error) {
 			r, err := buildUserRunner(cfg, st, secretStore, userID, log)
 			if err != nil {
 				return runner.Stats{}, err
 			}
 			return r.RunOnce(ctx)
 		}
+		// One lock shared by the scheduler and connect-triggered passes (#6) so
+		// they never fetch concurrently — the single-fetcher invariant (ADR-018).
+		runUser := serialize(&sync.Mutex{}, rawRunUser)
+		if app.Connect != nil {
+			app.Connect.Discovery = &discoveryTrigger{ctx: ctx, run: runUser, log: log}
+			log.Info("connect-triggered discovery enabled")
+		}
 		go runScheduler(ctx, cfg.DiscoveryInterval, st, runUser, log)
 	} else {
 		log.Info("scheduled discovery disabled (TAPIR_DISCOVERY_INTERVAL unset or 0)")