feat(discovery): per-channel caption-availability memory (ADR-024)
After transcript caching (ADR-021) and the Shorts filter (ADR-023), the remaining caption waste is the first fetch on every new video of a channel that never has English captions — each costs one rate-limited fetch to resolve to "none", and on a throttled IP churns the backoff machinery first. Remember, per (user, channel), a streak of consecutive no-caption outcomes (channel_caption_state, migration 016, RLS-scoped). Once it reaches TAPIR_CHANNEL_CAPTIONLESS_THRESHOLD (default 5) the channel is suppressed — videos discovered/listed but not caption-fetched — for TAPIR_CHANNEL_CAPTIONLESS_WINDOW (default 14d), then one is re-probed (auto-recovery). A successful fetch resets the streak; a 429 does not count; an explicit manual request bypasses suppression. threshold=0 disables. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,83 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// CaptionlessChannels returns the set of channel ids currently suppressed for the
|
||||
// user — channels whose recent videos all yielded no captions, within their
|
||||
// suppression window (ADR-024). The runner skips caption fetches for these
|
||||
// channels' videos. A channel whose window has expired is not returned, so its
|
||||
// next video is re-probed (auto-recovery).
|
||||
func (s *Store) CaptionlessChannels(ctx context.Context, userID string) (map[string]bool, error) {
|
||||
out := map[string]bool{}
|
||||
err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
|
||||
rows, err := tx.Query(ctx, `
|
||||
SELECT channel_id FROM channel_caption_state
|
||||
WHERE user_id = $1 AND captionless_until IS NOT NULL AND captionless_until > now()`,
|
||||
userID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: caption-less channels: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var ch string
|
||||
if err := rows.Scan(&ch); err != nil {
|
||||
return fmt.Errorf("store: scan caption-less channel: %w", err)
|
||||
}
|
||||
out[ch] = true
|
||||
}
|
||||
return rows.Err()
|
||||
})
|
||||
return out, err
|
||||
}
|
||||
|
||||
// RecordChannelCaptionOutcome updates a channel's caption-availability memory
|
||||
// after a fetch attempt (ADR-024). hadCaptions resets the channel (consecutive
|
||||
// count to 0, suppression cleared). Otherwise the consecutive no-caption count is
|
||||
// incremented; once it reaches threshold the channel is suppressed for window.
|
||||
// threshold <= 0 is a no-op (feature disabled). An empty channelID is ignored
|
||||
// (some sources may not carry one).
|
||||
func (s *Store) RecordChannelCaptionOutcome(ctx context.Context, userID, channelID string, hadCaptions bool, threshold int, window time.Duration) error {
|
||||
if channelID == "" || threshold <= 0 {
|
||||
return nil
|
||||
}
|
||||
return s.withUser(ctx, userID, func(tx pgx.Tx) error {
|
||||
if hadCaptions {
|
||||
_, err := tx.Exec(ctx, `
|
||||
INSERT INTO channel_caption_state (user_id, channel_id, consecutive_none, captionless_until, updated_at)
|
||||
VALUES ($1, $2, 0, NULL, now())
|
||||
ON CONFLICT (user_id, channel_id)
|
||||
DO UPDATE SET consecutive_none = 0, captionless_until = NULL, updated_at = now()`,
|
||||
userID, channelID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: reset channel caption state: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
// No captions: increment the streak; suppress once it reaches threshold.
|
||||
// captionless_until is set from the NEW count inside the same statement so
|
||||
// the decision is atomic with the increment.
|
||||
until := time.Now().Add(window)
|
||||
_, err := tx.Exec(ctx, `
|
||||
INSERT INTO channel_caption_state (user_id, channel_id, consecutive_none, captionless_until, updated_at)
|
||||
VALUES ($1, $2, 1, CASE WHEN 1 >= $3 THEN $4::timestamptz ELSE NULL END, now())
|
||||
ON CONFLICT (user_id, channel_id)
|
||||
DO UPDATE SET
|
||||
consecutive_none = channel_caption_state.consecutive_none + 1,
|
||||
captionless_until = CASE
|
||||
WHEN channel_caption_state.consecutive_none + 1 >= $3 THEN $4::timestamptz
|
||||
ELSE channel_caption_state.captionless_until
|
||||
END,
|
||||
updated_at = now()`,
|
||||
userID, channelID, threshold, until)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: record channel no-caption: %w", err)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
package store_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestChannelCaptionMemory_SuppressesAfterThreshold(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newStore(t)
|
||||
super := rawPool(t)
|
||||
resetDB(t, super)
|
||||
seedUser(t, super, userA)
|
||||
|
||||
const threshold = 3
|
||||
window := time.Hour
|
||||
|
||||
// Below threshold: not yet suppressed.
|
||||
require.NoError(t, s.RecordChannelCaptionOutcome(ctx, userA, "chanX", false, threshold, window))
|
||||
require.NoError(t, s.RecordChannelCaptionOutcome(ctx, userA, "chanX", false, threshold, window))
|
||||
got, err := s.CaptionlessChannels(ctx, userA)
|
||||
require.NoError(t, err)
|
||||
require.NotContains(t, got, "chanX", "2 < threshold 3: not suppressed yet")
|
||||
|
||||
// Crossing the threshold suppresses the channel.
|
||||
require.NoError(t, s.RecordChannelCaptionOutcome(ctx, userA, "chanX", false, threshold, window))
|
||||
got, err = s.CaptionlessChannels(ctx, userA)
|
||||
require.NoError(t, err)
|
||||
require.Contains(t, got, "chanX", "3 consecutive no-caption results suppress the channel")
|
||||
|
||||
// A successful caption fetch resets it.
|
||||
require.NoError(t, s.RecordChannelCaptionOutcome(ctx, userA, "chanX", true, threshold, window))
|
||||
got, err = s.CaptionlessChannels(ctx, userA)
|
||||
require.NoError(t, err)
|
||||
require.NotContains(t, got, "chanX", "a captioned video clears suppression")
|
||||
}
|
||||
|
||||
func TestChannelCaptionMemory_WindowExpiryReProbes(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newStore(t)
|
||||
super := rawPool(t)
|
||||
resetDB(t, super)
|
||||
seedUser(t, super, userA)
|
||||
|
||||
// A negative window means captionless_until lands in the past — modelling an
|
||||
// elapsed suppression window, which must make the channel eligible again.
|
||||
require.NoError(t, s.RecordChannelCaptionOutcome(ctx, userA, "chanY", false, 1, -time.Hour))
|
||||
got, err := s.CaptionlessChannels(ctx, userA)
|
||||
require.NoError(t, err)
|
||||
require.NotContains(t, got, "chanY", "an expired window re-enables the channel for a re-probe")
|
||||
}
|
||||
|
||||
func TestChannelCaptionMemory_DisabledThresholdIsNoOp(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newStore(t)
|
||||
super := rawPool(t)
|
||||
resetDB(t, super)
|
||||
seedUser(t, super, userA)
|
||||
|
||||
require.NoError(t, s.RecordChannelCaptionOutcome(ctx, userA, "chanZ", false, 0, time.Hour))
|
||||
got, err := s.CaptionlessChannels(ctx, userA)
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, got, "threshold 0 disables the memory — nothing recorded")
|
||||
}
|
||||
@@ -53,7 +53,9 @@ func TestMigration010LoginEventsUpDown(t *testing.T) {
|
||||
require.True(t, loginEventsExists(t), "login_events must exist at latest migration")
|
||||
|
||||
m := fileMigrator(t)
|
||||
// 011..015 sit above 010; step them down first so 010 is exercised in isolation.
|
||||
// 011..016 sit above 010; step them down first so 010 is exercised in isolation.
|
||||
require.NoError(t, m.Steps(-1), "down 016 drops channel_caption_state, login_events intact")
|
||||
require.True(t, loginEventsExists(t), "016 down leaves login_events intact")
|
||||
require.NoError(t, m.Steps(-1), "down 015 reshapes transcripts, login_events intact")
|
||||
require.True(t, loginEventsExists(t), "015 down leaves login_events intact")
|
||||
require.NoError(t, m.Steps(-1), "down 014 drops channel_title, login_events intact")
|
||||
@@ -68,7 +70,7 @@ func TestMigration010LoginEventsUpDown(t *testing.T) {
|
||||
require.NoError(t, m.Steps(-1), "down 010 must drop login_events")
|
||||
require.False(t, loginEventsExists(t), "login_events must be gone after the down migration")
|
||||
|
||||
require.NoError(t, m.Steps(6), "up must recreate 010 then re-apply 011..015")
|
||||
require.NoError(t, m.Steps(7), "up must recreate 010 then re-apply 011..016")
|
||||
require.True(t, loginEventsExists(t), "login_events must be restored after the up migration")
|
||||
}
|
||||
|
||||
@@ -91,6 +93,7 @@ func TestMigration011AutoSummarizeDefaultUpDown(t *testing.T) {
|
||||
require.Equal(t, "true", autoSummarizeDefault(t), "011 sets the default to TRUE")
|
||||
|
||||
m := fileMigrator(t)
|
||||
require.NoError(t, m.Steps(-1), "down 016 drops channel_caption_state")
|
||||
require.NoError(t, m.Steps(-1), "down 015 reshapes transcripts")
|
||||
require.NoError(t, m.Steps(-1), "down 014 drops channel_title")
|
||||
require.NoError(t, m.Steps(-1), "down 013 drops channel_errors")
|
||||
@@ -104,6 +107,7 @@ func TestMigration011AutoSummarizeDefaultUpDown(t *testing.T) {
|
||||
require.NoError(t, m.Steps(1), "up 013 creates channel_errors")
|
||||
require.NoError(t, m.Steps(1), "up 014 recreates channel_title")
|
||||
require.NoError(t, m.Steps(1), "up 015 reshapes transcripts to shared")
|
||||
require.NoError(t, m.Steps(1), "up 016 recreates channel_caption_state (HEAD)")
|
||||
}
|
||||
|
||||
// channelTitleExists reports whether videos.channel_title is present.
|
||||
@@ -123,6 +127,8 @@ func TestMigration014VideoChannelTitleUpDown(t *testing.T) {
|
||||
require.True(t, channelTitleExists(t), "channel_title exists at latest migration")
|
||||
|
||||
m := fileMigrator(t)
|
||||
require.NoError(t, m.Steps(-1), "down 016 drops channel_caption_state, channel_title intact")
|
||||
require.True(t, channelTitleExists(t), "016 down leaves channel_title intact")
|
||||
require.NoError(t, m.Steps(-1), "down 015 reshapes transcripts, channel_title intact")
|
||||
require.True(t, channelTitleExists(t), "015 down leaves channel_title intact")
|
||||
require.NoError(t, m.Steps(-1), "down 014 must drop channel_title")
|
||||
@@ -130,7 +136,8 @@ func TestMigration014VideoChannelTitleUpDown(t *testing.T) {
|
||||
|
||||
require.NoError(t, m.Steps(1), "up 014 must recreate channel_title")
|
||||
require.True(t, channelTitleExists(t), "channel_title must be restored after the up migration")
|
||||
require.NoError(t, m.Steps(1), "up 015 restores the shared transcripts shape (HEAD)")
|
||||
require.NoError(t, m.Steps(1), "up 015 restores the shared transcripts shape")
|
||||
require.NoError(t, m.Steps(1), "up 016 recreates channel_caption_state (HEAD)")
|
||||
}
|
||||
|
||||
// TestMigration012FixAutoSummarizeRLS proves 012 runs cleanly and flips any
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE channel_caption_state;
|
||||
@@ -0,0 +1,30 @@
|
||||
-- Migration 016: per-(user, channel) caption-availability memory (ADR-024).
|
||||
--
|
||||
-- Some channels never publish English captions (foreign-language news, music,
|
||||
-- etc.). Each of their new videos still costs ONE rate-limited caption fetch
|
||||
-- (ADR-014) before resolving to "none" — and on a throttled egress IP that fetch
|
||||
-- may 429 and churn through the backoff machinery first. This table remembers
|
||||
-- channels that repeatedly yield no captions so discovery can stop attempting
|
||||
-- their videos, freeing the scarce fetch budget for channels that do have them.
|
||||
--
|
||||
-- consecutive_none counts no-caption outcomes in a row; a successful fetch resets
|
||||
-- it to 0. Once it crosses the threshold the channel is suppressed until
|
||||
-- captionless_until, after which one video is re-probed (auto-recovery for a
|
||||
-- channel that starts adding captions). Per-user + RLS-scoped, consistent with
|
||||
-- the rest of the user-owned schema (subscriptions are per-user; ADR-012).
|
||||
CREATE TABLE channel_caption_state (
|
||||
user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
channel_id TEXT NOT NULL,
|
||||
consecutive_none INT NOT NULL DEFAULT 0,
|
||||
captionless_until TIMESTAMPTZ,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (user_id, channel_id)
|
||||
);
|
||||
|
||||
CREATE INDEX idx_channel_caption_state_user_id ON channel_caption_state(user_id);
|
||||
|
||||
ALTER TABLE channel_caption_state ENABLE ROW LEVEL SECURITY;
|
||||
ALTER TABLE channel_caption_state FORCE ROW LEVEL SECURITY;
|
||||
CREATE POLICY channel_caption_state_isolation ON channel_caption_state
|
||||
FOR ALL
|
||||
USING (user_id = current_setting('tapir.current_user_id', true)::uuid);
|
||||
Reference in New Issue
Block a user