Pipeline v2 core: platform-authoritative drop state (opt-in)

AGENTS.md

@@ -63,10 +63,13 @@ proto/                  buf config + gRPC defs for the sidecar — core, not dea
 
 ## Platform gotchas
 
-- **Twitch = Android device-code OAuth over direct HTTP.** Never send a
-  `Client-Integrity` header — adding it is a self-inflicted integrity wall. Do
-  not re-implement logged-out catalog scraping; `/drops/campaigns` while logged
-  out is just a login wall.
+- **Twitch clients: Android (legacy sessions) and TV (device-code logins since
+  v1.4.0) over direct HTTP.** Never send a `Client-Integrity` header on the
+  Android or TV client; adding it there is a self-inflicted integrity wall. The
+  `Client-Integrity` header is allowed ONLY on the web-client profile
+  (browser-captured sessions), where Twitch requires it. Do not re-implement
+  logged-out catalog scraping; `/drops/campaigns` while logged out is just a
+  login wall.
 - **Twitch credit beacon is gated by `HeartbeatInterval`** (1 minute credited
   per beacon), so anything above 60s under-credits. Keep it at 60s. Kick is
   immune.

CLAUDE.md

@@ -57,8 +57,11 @@ the emoji-section + bold-subject format above.
 
 ## Platform gotchas
 
-- **Twitch = Android device-code OAuth over direct HTTP.** Never send a
-  `Client-Integrity` header; adding it was a self-inflicted integrity wall.
+- **Twitch clients: Android (legacy sessions) and TV (device-code logins since
+  v1.4.0) over direct HTTP.** Never send a `Client-Integrity` header on the
+  Android or TV client; adding it there was a self-inflicted integrity wall.
+  The `Client-Integrity` header is allowed ONLY on the web-client profile
+  (browser-captured sessions, rangermix-style), where Twitch requires it.
   Don't re-implement logged-out catalog scraping either — `/drops/campaigns`
   while logged out is just a login wall.
 - **Twitch credit beacon is gated by `HeartbeatInterval`** (1 minute credited

README.md

@@ -1,3 +1,7 @@
+> [!WARNING]
+> **Drop discovery is limited.** 
+Twitch no longer lists campaigns for new logins, so GrubDrops only finds drops for games on your whitelist.
+
 <p align="center">
   <img src="internal/web/static/img/logo.png" width="160" alt="GrubDrops">
 </p>
@@ -221,6 +225,7 @@ default shown.
 | `GRUB_AUTHBYPASS` | `false` | **Disables all auth** when truthy (`1`/`true`). |
 | `GRUB_TWITCH_BROWSER` | `0` | `1` routes Twitch through the browser sidecar instead of direct HTTP. Experimental; the default direct-HTTP path is recommended. |
 | `GRUB_CANARY_INTERVAL` | Health-tab value | Overrides the accrual-canary run cadence (e.g. `6h`); falls back to the Settings ▸ Health value. |
+| `GRUB_PIPELINE` | `v2` | `v1` falls back to the legacy watcher process-wide; also overridable per account (kv `pipeline_override:<accountID>`). |
 
 > **"Invalid CSRF token"?** `GRUB_SECURE_COOKIES` must match your scheme: `0`
 > over plain HTTP, `1` over HTTPS (proxy must forward `X-Forwarded-Proto: https`).

cmd/miner/main.go

@@ -31,6 +31,7 @@ import (
 	mlog "github.com/aalejandrofer/grubdrops/internal/log"
 	"github.com/aalejandrofer/grubdrops/internal/netutil"
 	"github.com/aalejandrofer/grubdrops/internal/notify"
+	"github.com/aalejandrofer/grubdrops/internal/pipeline/loop"
 	"github.com/aalejandrofer/grubdrops/internal/platform"
 	"github.com/aalejandrofer/grubdrops/internal/platform/kick"
 	"github.com/aalejandrofer/grubdrops/internal/platform/twitch"
@@ -252,6 +253,11 @@ func run() error {
 	} else {
 		twitchBackend = twitch.New()
 	}
+	// One shared Twitch campaign catalog for the whole process: backends on
+	// the Android/legacy login publish the full campaign list, TV-login
+	// backends merge the campaigns Twitch hides from the TV client.
+	twitchCatalog := twitch.NewCatalog()
+	twitchBackend.SetCatalog(twitchCatalog)
 	if twitchBrowserEnabled && browserClient != nil {
 		// proxyTransport is nil when no proxy is configured; the browser
 		// backend then dials the Spade beacon direct, same as before.
@@ -336,6 +342,16 @@ func run() error {
 	// and the /drops Past + /history views stay empty.
 	claimRecorder := store.NewClaimRecorder(q)
 
+	// dropStore is the pipeline v2 per-account drop_state table. Backfill
+	// seeds it from the v1 in-progress/claims state on every boot so a
+	// GRUB_PIPELINE=v2 opt-in mid-lifecycle doesn't start from scratch.
+	dropStore := store.NewDropStateStore(q)
+	if n, err := store.BackfillDropState(ctx, q, time.Now()); err != nil {
+		logger.Warn("pipeline v2: drop_state backfill failed", "err", err)
+	} else if n > 0 {
+		logger.Info("pipeline v2: drop_state backfilled from v1 state", "rows", n)
+	}
+
 	// Per-account direct-Twitch backends. The direct twitch.Backend holds
 	// per-account state (auth, userID/userLogin caches, its own PubSub
 	// socket), so sharing ONE instance across accounts races their tokens
@@ -367,6 +383,7 @@ func run() error {
 			} else {
 				bk = twitch.New()
 			}
+			bk.SetCatalog(twitchCatalog)
 			twitchBackends[a.ID] = bk
 			return bk, true
 		}
@@ -461,6 +478,36 @@ func run() error {
 		forceCollected := loadCollectOverrides(ctx, q)
 		persistedSkips, skipRecorder, skipClearer := loadSkipOverrides(ctx, q)
 
+		if pipelineModeFor(ctx, q, a.ID) == "v2" {
+			prio, err := dropStore.StreamerPriority(ctx, a.ID)
+			if err != nil {
+				logger.Warn("pipeline v2: load streamer priority failed", "account", a.ID, "err", err)
+			}
+			force := v2ForceChannels(ctx, q, a.ID)
+			l, err := loop.New(loop.Config{
+				AccountID: a.ID, AccountLabel: a.DisplayName, Platform: a.Platform,
+				Backend: b, Session: sess,
+				Store: dropStore, History: claimRecorder, Persister: campaignPersister,
+				Notifier:  notifier,
+				AllowGame: allow, AllowChannel: matchAnyChannel(prio), GameRank: rank,
+				Games:                 names,
+				PriorityMode:          priorityMode,
+				ForceLinked:           forceLinked,
+				StreamerPriority:      prio,
+				ForceWatch:            force,
+				ProgressNotifyStepPct: progressStep,
+				// Twitch hides drops claimed outside the app once their
+				// campaign leaves the Inventory; Kick keeps them listed.
+				ClaimProbe:      a.Platform == "twitch",
+				EnrollDiscovery: enrollDiscoveryFor(a.Platform, sess),
+			})
+			if err == nil {
+				logger.Info("pipeline v2 enabled for account", "account", a.ID, "platform", a.Platform)
+				return scheduler.NewEntry(a.ID, l), nil
+			}
+			logger.Warn("pipeline v2 unavailable, falling back to v1", "account", a.ID, "err", err)
+		}
+
 		acctLabel := a.DisplayName
 		w := watcher.New(watcher.Config{
 			AccountID: a.ID, AccountLabel: acctLabel, Platform: a.Platform,
@@ -908,6 +955,52 @@ func decodeKickChannels(s platform.Session) []string {
 	return out
 }
 
+// pipelineModeFor picks v1 or v2 for an account: kv override first, then
+// GRUB_PIPELINE, default v2. Set GRUB_PIPELINE=v1 (or a per-account
+// override) to fall back to the legacy watcher; build() also falls back
+// automatically per-account when loop.New itself fails (e.g. no
+// BrowserBackend wired for Kick).
+func pipelineModeFor(ctx context.Context, q *gen.Queries, accountID string) string {
+	if v, err := q.GetSettingString(ctx, store.PipelineOverridePrefix+accountID); err == nil {
+		if s := string(v); s == "v1" || s == "v2" {
+			return s
+		}
+	}
+	if os.Getenv("GRUB_PIPELINE") == "v1" {
+		return "v1"
+	}
+	return "v2"
+}
+
+// enrollDiscoveryFor turns on pipeline v2 watch-to-enroll discovery for
+// TV-client Twitch sessions: they can't read the campaign list, so they
+// find campaigns by watching a drops-enabled channel per whitelisted game
+// until Inventory lists what Twitch enrolled them in. Android sessions see
+// the list directly and Kick has no such gap, so both stay off.
+func enrollDiscoveryFor(platformName string, sess platform.Session) bool {
+	return platformName == "twitch" && sess.ClientID == twitch.ClientTV
+}
+
+// matchAnyChannel reports whether any campaign channel is one of logins.
+// Nil when logins is empty, so the null-game gate stays off.
+func matchAnyChannel(logins []string) func([]string) bool {
+	if len(logins) == 0 {
+		return nil
+	}
+	set := make(map[string]bool, len(logins))
+	for _, l := range logins {
+		set[strings.ToLower(l)] = true
+	}
+	return func(chs []string) bool {
+		for _, c := range chs {
+			if set[strings.ToLower(c)] {
+				return true
+			}
+		}
+		return false
+	}
+}
+
 // loadAccountWhitelist materialises the per-account game allow-list
 // into match + rank closures the watcher consumes, plus the plain
 // display names (as stored in the games table) that feed
@@ -1006,8 +1099,7 @@ func loadAccountChannels(ctx context.Context, q *gen.Queries, accountID string)
 type forceWatchStore struct{ q *gen.Queries }
 
 func (f forceWatchStore) Next(ctx context.Context, accountID string) (watcher.ForceTask, bool) {
-	v, err := f.q.GetSettingString(ctx, api.ForceWatchEnabledKey(accountID))
-	if err != nil || string(v) != "1" {
+	if !forceWatchEnabled(ctx, f.q, accountID) {
 		return watcher.ForceTask{}, false
 	}
 	rows, err := f.q.ListForceChannels(ctx, accountID)
@@ -1017,6 +1109,35 @@ func (f forceWatchStore) Next(ctx context.Context, accountID string) (watcher.Fo
 	return watcher.ForceTask{Channel: rows[0].Channel}, true
 }
 
+// forceWatchEnabled reports whether the per-account force-watch toggle
+// (force_watch:<accountID> KV flag, set via /accounts/:id/force-watch) is
+// on. Shared by v1's forceWatchStore and the v2 wiring so both agree on
+// when a configured force-watch channel actually applies.
+func forceWatchEnabled(ctx context.Context, q *gen.Queries, accountID string) bool {
+	v, err := q.GetSettingString(ctx, api.ForceWatchEnabledKey(accountID))
+	return err == nil && string(v) == "1"
+}
+
+// v2ForceChannels returns the account's configured force-watch channel
+// logins when the force-watch toggle is enabled, or nil otherwise. Used by
+// the pipeline v2 branch in build() so a disabled toggle behaves like v1's
+// forceWatchStore.Next (no forced channel), instead of always passing the
+// configured list.
+func v2ForceChannels(ctx context.Context, q *gen.Queries, accountID string) []string {
+	if !forceWatchEnabled(ctx, q, accountID) {
+		return nil
+	}
+	rows, err := q.ListForceChannels(ctx, accountID)
+	if err != nil {
+		return nil
+	}
+	var force []string
+	for _, r := range rows {
+		force = append(force, r.Channel)
+	}
+	return force
+}
+
 // parseDuration parses a Go duration string (e.g. "5m", "30s") with a
 // fallback when the env var is empty or malformed. Matches the pattern
 // used elsewhere in the codebase for config-by-env.

cmd/miner/main_test.go

@@ -8,10 +8,22 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/aalejandrofer/grubdrops/internal/api"
+	"github.com/aalejandrofer/grubdrops/internal/platform"
+	"github.com/aalejandrofer/grubdrops/internal/platform/twitch"
 	"github.com/aalejandrofer/grubdrops/internal/store"
 	"github.com/aalejandrofer/grubdrops/internal/store/gen"
 )
 
+// TestEnrollDiscoveryFor: watch-to-enroll discovery is for TV-client Twitch
+// sessions only; Android (legacy empty ClientID) and Kick stay off.
+func TestEnrollDiscoveryFor(t *testing.T) {
+	assert.True(t, enrollDiscoveryFor("twitch", platform.Session{ClientID: twitch.ClientTV}))
+	assert.False(t, enrollDiscoveryFor("twitch", platform.Session{}), "Android / legacy session")
+	assert.False(t, enrollDiscoveryFor("twitch", platform.Session{ClientID: "android"}))
+	assert.False(t, enrollDiscoveryFor("kick", platform.Session{ClientID: twitch.ClientTV}))
+}
+
 func TestLoadAccountChannels(t *testing.T) {
 	ctx := context.Background()
 	db, err := store.Open(ctx, t.TempDir()+"/t.db")
@@ -64,3 +76,76 @@ func TestGenerateMasterKey_IsAcceptedByCryptor(t *testing.T) {
 	require.NoError(t, err)
 	assert.NotEqual(t, key, key2, "each keygen must be unique")
 }
+
+func TestPipelineModeFor(t *testing.T) {
+	ctx := context.Background()
+	db, err := store.Open(ctx, t.TempDir()+"/t.db")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = db.Close() })
+	q := gen.New(db)
+
+	t.Setenv("GRUB_PIPELINE", "")
+	assert.Equal(t, "v2", pipelineModeFor(ctx, q, "acc"), "v2 is the default")
+	t.Setenv("GRUB_PIPELINE", "v1")
+	assert.Equal(t, "v1", pipelineModeFor(ctx, q, "acc"), "GRUB_PIPELINE=v1 falls back to the legacy watcher")
+	require.NoError(t, q.UpsertSettingString(ctx, gen.UpsertSettingStringParams{Key: store.PipelineOverridePrefix + "acc", Value: []byte("v2")}))
+	assert.Equal(t, "v2", pipelineModeFor(ctx, q, "acc"), "per-account override beats env")
+	require.NoError(t, q.UpsertSettingString(ctx, gen.UpsertSettingStringParams{Key: store.PipelineOverridePrefix + "acc", Value: []byte("junk")}))
+	assert.Equal(t, "v1", pipelineModeFor(ctx, q, "acc"), "invalid override falls through to env (still v1 here)")
+}
+
+func TestMatchAnyChannel(t *testing.T) {
+	assert.Nil(t, matchAnyChannel(nil))
+	m := matchAnyChannel([]string{"Fav"})
+	assert.True(t, m([]string{"x", "fav"}))
+	assert.False(t, m([]string{"x"}))
+}
+
+// TestForceWatchEnabled proves the extracted helper reads the same
+// force_watch:<accountID> KV flag forceWatchStore.Next checks, so v1 and v2
+// agree on whether force-watch is on for an account.
+func TestForceWatchEnabled(t *testing.T) {
+	ctx := context.Background()
+	db, err := store.Open(ctx, t.TempDir()+"/t.db")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = db.Close() })
+	q := gen.New(db)
+
+	assert.False(t, forceWatchEnabled(ctx, q, "acc-1"), "flag unset -> disabled")
+
+	require.NoError(t, q.UpsertSettingString(ctx, gen.UpsertSettingStringParams{
+		Key: api.ForceWatchEnabledKey("acc-1"), Value: []byte("1"),
+	}))
+	assert.True(t, forceWatchEnabled(ctx, q, "acc-1"), "flag=1 -> enabled")
+
+	assert.False(t, forceWatchEnabled(ctx, q, "acc-2"), "different account unaffected")
+}
+
+// TestV2ForceChannels proves the v2 wiring only surfaces ListForceChannels
+// rows when the account's force-watch flag is enabled — matching v1's
+// forceWatchStore.Next gate (previously v2 passed the list unconditionally).
+func TestV2ForceChannels(t *testing.T) {
+	ctx := context.Background()
+	db, err := store.Open(ctx, t.TempDir()+"/t.db")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = db.Close() })
+	q := gen.New(db)
+
+	now := time.Now().Unix()
+	_, err = q.CreateAccount(ctx, gen.CreateAccountParams{
+		ID: "acc-1", Platform: "kick", DisplayName: "k",
+		Status: "idle", FingerprintJson: "{}", Enabled: 1,
+		CreatedAt: now, UpdatedAt: now,
+	})
+	require.NoError(t, err)
+	require.NoError(t, q.AddForceChannel(ctx, gen.AddForceChannelParams{
+		AccountID: "acc-1", Channel: "somechannel", Rank: 0, CreatedAt: now,
+	}))
+
+	assert.Empty(t, v2ForceChannels(ctx, q, "acc-1"), "flag off -> no force channels even though rows exist")
+
+	require.NoError(t, q.UpsertSettingString(ctx, gen.UpsertSettingStringParams{
+		Key: api.ForceWatchEnabledKey("acc-1"), Value: []byte("1"),
+	}))
+	assert.Equal(t, []string{"somechannel"}, v2ForceChannels(ctx, q, "acc-1"), "flag on -> rows surfaced")
+}

docs/CHANGELOG.md

@@ -4,6 +4,57 @@ All notable changes to GrubDrops.
 
 ## [Unreleased]
 
+### Added
+
+- Pipeline v2 (opt-in, `GRUB_PIPELINE=v2` or kv `pipeline_override:<account>`): drop state now comes from each campaign's platform details, so drops claimed on the website or with a lost claim response show as claimed and mining moves on (#50).
+- Streamer priority for pipeline v2: per-account priority streamers are watched first, then any live channel in the category (#49). Seeded from the existing null-game channel list. Until the settings editor lands, v2 reads a one-time copy of the null-game channel list taken at upgrade.
+- `drop_state` and `account_streamer_priority` tables (migration 0016), with a one-time backfill from claim history, manual marks and ghost-skips.
+- Accounts on the Twitch TV login share the full campaign list found by any Android-login account in the same GrubDrops, so campaigns hidden from the TV client (for example behind a global badge campaign) are still found.
+- A banner explains that Twitch limits campaign discovery for new (TV) logins and points to the game whitelist.
+- Accounts on the Twitch TV login find campaigns by briefly watching a top drops-enabled channel for each whitelisted game when there is nothing else to mine, so Twitch enrolls them and the campaigns appear. (Pipeline v2: 10 minutes per game, each game at most once every 6 hours; dashboard state `discovering`.) (the dashboard shows these accounts as Discovering)
+
+### Changed
+
+- Twitch claim code split into status fetch + classification (behaviour unchanged).
+- Scheduler reads dashboard snapshots and discoveries through interfaces, so v1 and v2 runners both report state.
+- Pipeline v2 is now the default. Set `GRUB_PIPELINE=v1` (or a per-account override) to fall back to the legacy watcher.
+- Reloading accounts builds them in parallel, so a slow account no longer holds up the others.
+
+### Fixed
+
+- Streamers' own channel campaigns with no game no longer flood the drops page, and campaigns past their end time show as ended.
+- Twitch TV-channel discovery now matches a campaign's game by ID first (falling back to the display-name slug only when either ID is missing), so a campaign whose display name doesn't match the live directory's is no longer wrongly skipped.
+- The shared Twitch discovery catalog no longer accumulates stale entries from recycled account backends; sources older than the TTL are evicted, not just ignored.
+- Idle Kick accounts no longer flood the log with state changes every few seconds.
+
+### Fixed (pipeline v2 only)
+
+- Claim failures back off 1m, 5m, 30m and stop after 5 attempts instead of looping forever.
+- Kick link-required claims block with a reason instead of retrying every poll.
+- Kick required minutes come from the platform, not an invented 120.
+- Ghost-skipped drops are re-checked instead of being skipped forever.
+- Migrated Kick ghost-skips pick up the campaign's required minutes instead of staying blocked as sub-only; backfilled rows now carry required minutes.
+- Manual "mark collected" marks migrate as user claims, not final platform claims.
+- A campaign link block lifts on the next sync after the user links, instead of holding for 24h.
+- Twitch priority streamers are not probed for restricted campaigns whose allow-list is unknown, so v2 never watches a channel the campaign doesn't credit.
+- Twitch drops with no viewer record are no longer read as a definite "not enrolled".
+- The Twitch integrity wall shows the account as needing re-auth and stops the loop, like v1.
+- Session restarts wait for the previous watch to stop, and a restart on the same channel keeps its PubSub subscription.
+- A drop that falls back from claimable resets its claim failure count.
+- v2 discovers campaigns for TV-client Twitch accounts (the whitelisted games now reach the session).
+- v2 reads TV-account Twitch progress from the inventory, and sends every v2 request under the account's own Twitch client.
+- v2 no longer mines subscription-only Twitch drops.
+- A Twitch drop claimed outside GrubDrops whose campaign left the inventory is detected with one claim check when watching it stalls, instead of being watched again and again.
+- v2 heartbeats appear in the dashboard events feed, like v1's.
+- The stall claim check only runs on a drop's second stall, and an empty Twitch claim status counts as a failure, so a blind check can never mark an unenrolled drop claimed.
+- Drops already claimed under v1 no longer look unclaimed to v2 when an account switches over; v2 reads the claim history after every sync, not only at startup.
+- v2 no longer watches drops from campaigns that need an unlinked account between syncs.
+- Channel whitelist edits on the drops page now reach v2 accounts.
+- The force-watch on/off switch now applies to v2 accounts.
+- A newly added Twitch account that already finished its drops is recognised within a few syncs instead of being watched drop by drop.
+- A Twitch claim notification no longer overrides a drop you skipped or that needs a linked account.
+- v2 no longer bounces between two channels every few minutes when a channel drops out of Twitch's directory list.
+
 ## [1.4.2] — 2026-09-28
 
 ### Fixed