diff --git a/cmd/tapir/discovery.go b/cmd/tapir/discovery.go new file mode 100644 index 0000000..7d5cf7b --- /dev/null +++ b/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) + } + }() +} diff --git a/cmd/tapir/discovery_test.go b/cmd/tapir/discovery_test.go new file mode 100644 index 0000000..553f485 --- /dev/null +++ b/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") + } +} diff --git a/cmd/tapir/main.go b/cmd/tapir/main.go index 7dc2716..4b8ed23 100644 --- a/cmd/tapir/main.go +++ b/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)") diff --git a/docs/use-cases/connect_account.feature b/docs/use-cases/connect_account.feature index 31af737..498a647 100644 --- a/docs/use-cases/connect_account.feature +++ b/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 diff --git a/internal/web/connect.go b/internal/web/connect.go index 5d89f87..2dade16 100644 --- a/internal/web/connect.go +++ b/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) } diff --git a/internal/web/connect_test.go b/internal/web/connect_test.go index 50a7d68..7098113 100644 --- a/internal/web/connect_test.go +++ b/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}`) diff --git a/test/acceptance/scenario_coverage_test.go b/test/acceptance/scenario_coverage_test.go index d60146b..1ae8d8b 100644 --- a/test/acceptance/scenario_coverage_test.go +++ b/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",