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>

CLAUDE.md

@@ -85,7 +85,7 @@ Skills live in the canonical library `mathias/skills` and are wired into this re
 
 ## Current build state (start here for the first task)
 
-The repo is **green and shipping** — last tag `v0.8.0`. `task check` passes (fmt, vet, lint,
+The repo is **green and shipping** — last tag `v0.9.0`. `task check` passes (fmt, vet, lint,
 `go test -p 1 ./...`). Go is `1.26.1` (see `go.mod`).
 
 - Clean Architecture core is implemented: `internal/domain` (entities), `internal/ports`

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

docs/use-cases/connect_account.feature

@@ -18,6 +18,12 @@ Feature: Connect and manage video accounts
     Then the connection is stored with status "active"
     And my subscriptions are synced
 
+  Scenario: Connecting an account discovers videos immediately
+    Given I have no connected video accounts
+    When I connect my YouTube account
+    Then a discovery pass for my account is triggered right away
+    And I do not have to wait for the next scheduled pass to see my videos
+
   Scenario: Tokens are never stored in the clear
     When I connect any video account
     Then no OAuth token value is stored in the database

internal/web/connect.go

@@ -23,6 +23,15 @@ type Connections interface {
 	UpsertConnection(ctx context.Context, userID string, c store.Connection) error
 }
 
+// DiscoveryTrigger requests an out-of-band discovery pass for a user. The connect
+// flow fires it the moment a YouTube account is linked so videos appear promptly
+// instead of waiting for the next scheduled pass (#6). Enqueue must be
+// non-blocking and safe to call from the request goroutine; the implementation
+// owns serialization with the scheduler (one pass at a time). nil = no trigger.
+type DiscoveryTrigger interface {
+	Enqueue(userID string)
+}
+
 // connectStateTTL bounds how long a generated CSRF state is valid between the
 // connect redirect and the provider callback.
 const connectStateTTL = 10 * time.Minute
@@ -43,6 +52,10 @@ type ConnectHandler struct {
 	Conns   Connections
 	Log     *slog.Logger
 
+	// Discovery, when set, is fired after a successful connect so the new
+	// connection's videos are discovered immediately (#6). Optional.
+	Discovery DiscoveryTrigger
+
 	states *connectStateStore
 	now    func() time.Time
 }
@@ -134,6 +147,12 @@ func (h *ConnectHandler) handleCallback(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
+	// Discover this user's videos now rather than waiting for the next scheduled
+	// pass (#6). Non-blocking; the trigger serializes with the scheduler.
+	if h.Discovery != nil {
+		h.Discovery.Enqueue(userID)
+	}
+
 	setFlash(w, flashConnected)
 	http.Redirect(w, r, "/", http.StatusSeeOther)
 }

internal/web/connect_test.go

@@ -45,6 +45,24 @@ func (c *fakeConns) UpsertConnection(_ context.Context, userID string, conn stor
 	return nil
 }
 
+// fakeTrigger records Enqueue calls so a test can assert connect fired discovery.
+type fakeTrigger struct {
+	mu    sync.Mutex
+	users []string
+}
+
+func (f *fakeTrigger) Enqueue(userID string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.users = append(f.users, userID)
+}
+
+func (f *fakeTrigger) seen() []string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]string(nil), f.users...)
+}
+
 // tokenServer fakes Google's token endpoint, returning body for any POST.
 func tokenServer(t *testing.T, body string) *httptest.Server {
 	t.Helper()
@@ -126,6 +144,36 @@ func TestCallbackExchangesAndRecordsConnection(t *testing.T) {
 	require.Equal(t, wantRef, conns.conn.TokenRef)
 }
 
+func TestCallbackTriggersDiscovery(t *testing.T) {
+	srv := tokenServer(t,
+		`{"access_token":"at","refresh_token":"rt-secret","token_type":"Bearer","expires_in":3600}`)
+	app := newConnectApp(t, srv.URL, &fakeWriter{}, &fakeConns{})
+	trig := &fakeTrigger{}
+	app.Connect.Discovery = trig
+
+	state := connectState(t, app)
+	rec := do(t, app, httptest.NewRequest(http.MethodGet,
+		"/oauth/youtube/callback?state="+state+"&code=the-code", nil))
+
+	require.Equal(t, http.StatusSeeOther, rec.Code)
+	require.Equal(t, []string{userID}, trig.seen(),
+		"a successful connect must trigger discovery for the connecting user")
+}
+
+func TestCallbackNoDiscoveryOnFailedConnect(t *testing.T) {
+	srv := tokenServer(t,
+		`{"access_token":"at","refresh_token":"rt","token_type":"Bearer","expires_in":3600}`)
+	app := newConnectApp(t, srv.URL, &fakeWriter{}, &fakeConns{})
+	trig := &fakeTrigger{}
+	app.Connect.Discovery = trig
+
+	// No state → CSRF reject → nothing connected, so no discovery.
+	rec := do(t, app, httptest.NewRequest(http.MethodGet,
+		"/oauth/youtube/callback?code=the-code", nil))
+	require.Equal(t, http.StatusBadRequest, rec.Code)
+	require.Empty(t, trig.seen(), "a failed connect must not trigger discovery")
+}
+
 func TestCallbackRejectsMissingState(t *testing.T) {
 	srv := tokenServer(t,
 		`{"access_token":"at","refresh_token":"rt","token_type":"Bearer","expires_in":3600}`)

test/acceptance/scenario_coverage_test.go

@@ -38,6 +38,7 @@ var scenarioCoverage = map[string]string{
 
 	// connect_account.feature
 	"Connect a YouTube account":                              "TestCallbackExchangesAndRecordsConnection",
+	"Connecting an account discovers videos immediately":     "TestCallbackTriggersDiscovery",
 	"Tokens are never stored in the clear":                   "TestCallbackExchangesAndRecordsConnection",
 	"Revoking a connection stops watching but keeps history": "TestDisconnectRemovesTokenAndConnectionKeepsAccount",