diff --git a/internal/adapters/store/reads.go b/internal/adapters/store/reads.go index bd0b742..74a62d1 100644 --- a/internal/adapters/store/reads.go +++ b/internal/adapters/store/reads.go @@ -49,6 +49,10 @@ type SummaryRow struct { // set by the web "Summarize" button and cleared by the next `tapir run`. Only // populated by ListVideos/GetVideoRow (summary-only reads leave it false). SummarizeRequested bool + // TranscriptStatus mirrors videos.transcript_status (migration 007): "" (unset), + // "none", "rate_limited", or "fetched". Drives the "Retrying later" list badge. + // Only populated by ListVideos/GetVideoRow ("" on summary-only reads). + TranscriptStatus string } // selectSummary is the shared projection for both reads. videos is LEFT JOINed @@ -132,7 +136,8 @@ const selectVideo = ` COALESCE(s.fallback_used, FALSE), COALESCE(s.created_at, v.seen_at), (s.id IS NOT NULL) AS summarized, - v.summarize_requested + v.summarize_requested, + COALESCE(v.transcript_status, '') FROM videos v LEFT JOIN summaries s ON s.video_id = v.id AND s.user_id = v.user_id` @@ -245,6 +250,7 @@ func scanVideoRow(rows pgx.Row) (SummaryRow, error) { &row.CreatedAt, &row.Summarized, &row.SummarizeRequested, + &row.TranscriptStatus, ); err != nil { return SummaryRow{}, fmt.Errorf("store: scan video: %w", err) } diff --git a/internal/adapters/store/transcript_status.go b/internal/adapters/store/transcript_status.go new file mode 100644 index 0000000..c9cc369 --- /dev/null +++ b/internal/adapters/store/transcript_status.go @@ -0,0 +1,102 @@ +package store + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" +) + +// validTranscriptStatuses bounds SetTranscriptStatus input. "" clears the status +// (column NULL); the three named states mirror migration 007's documented values. +var validTranscriptStatuses = map[string]bool{ + "": true, + "none": true, + "rate_limited": true, + "fetched": true, +} + +// SetTranscriptStatus records the outcome of the last transcript attempt for a +// video (migration 007). When status is "rate_limited" it also stamps +// rate_limited_at = NOW() so the runner can back off; every other status clears +// that timestamp. "" unsets the status (column NULL). An unknown status is +// rejected. Scoped via withUser, so RLS confines the UPDATE to the caller's own +// video; ErrNotFound when the user has no such video. +func (s *Store) SetTranscriptStatus(ctx context.Context, userID, videoID, status string) error { + if !validTranscriptStatuses[status] { + return fmt.Errorf("store: invalid transcript status %q", status) + } + return s.withUser(ctx, userID, func(tx pgx.Tx) error { + ct, err := tx.Exec(ctx, + `UPDATE videos + SET transcript_status = NULLIF($1, ''), + rate_limited_at = CASE WHEN $1 = 'rate_limited' THEN NOW() ELSE NULL END + WHERE id = $2`, + status, videoID) + if err != nil { + return fmt.Errorf("store: set transcript status: %w", err) + } + if ct.RowsAffected() == 0 { + return ErrNotFound + } + return nil + }) +} + +// GetTranscriptStatus returns a video's transcript_status ("" when unset/NULL). +// Returns ErrNotFound when the user has no such video. Scoped via withUser. +func (s *Store) GetTranscriptStatus(ctx context.Context, userID, videoID string) (string, error) { + var status string + if err := s.withUser(ctx, userID, func(tx pgx.Tx) error { + err := tx.QueryRow(ctx, + `SELECT COALESCE(transcript_status, '') FROM videos WHERE id = $1`, videoID).Scan(&status) + if errors.Is(err, pgx.ErrNoRows) { + return ErrNotFound + } + return err + }); err != nil { + if errors.Is(err, ErrNotFound) { + return "", ErrNotFound + } + return "", fmt.Errorf("store: get transcript status: %w", err) + } + return status, nil +} + +// RateLimitedVideoIDs returns the user's videos currently in the "rate_limited" +// state, mapped to when the 429 was stamped (rate_limited_at). The run loop loads +// it once per pass (mirroring SeenVideoIDs) to skip re-fetching a video still +// inside the backoff window, saving caption requests. Scoped by user_id. +func (s *Store) RateLimitedVideoIDs(ctx context.Context, userID string) (map[string]time.Time, error) { + out := make(map[string]time.Time) + if err := s.withUser(ctx, userID, func(tx pgx.Tx) error { + rows, err := tx.Query(ctx, + `SELECT id, rate_limited_at FROM videos + WHERE user_id = $1 AND transcript_status = 'rate_limited' AND rate_limited_at IS NOT NULL`, + userID) + if err != nil { + return fmt.Errorf("store: rate limited video ids: %w", err) + } + defer rows.Close() + + for rows.Next() { + var ( + id string + at time.Time + ) + if err := rows.Scan(&id, &at); err != nil { + return fmt.Errorf("store: scan rate limited id: %w", err) + } + out[id] = at + } + if err := rows.Err(); err != nil { + return fmt.Errorf("store: iterate rate limited ids: %w", err) + } + return nil + }); err != nil { + return nil, err + } + return out, nil +} diff --git a/internal/adapters/store/transcript_status_test.go b/internal/adapters/store/transcript_status_test.go new file mode 100644 index 0000000..50b9edb --- /dev/null +++ b/internal/adapters/store/transcript_status_test.go @@ -0,0 +1,80 @@ +package store_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + "gitea.d-ma.be/mathias/tapir/internal/adapters/store" +) + +func TestSetTranscriptStatus_RoundTrip(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + + id, err := s.UpsertVideo(ctx, ytVideo(userA, "rt12345", "round trip")) + require.NoError(t, err) + + // Unset by default. + got, err := s.GetTranscriptStatus(ctx, userA, id) + require.NoError(t, err) + require.Equal(t, "", got) + + for _, status := range []string{"none", "fetched", "rate_limited", ""} { + require.NoError(t, s.SetTranscriptStatus(ctx, userA, id, status)) + got, err := s.GetTranscriptStatus(ctx, userA, id) + require.NoError(t, err) + require.Equal(t, status, got) + } +} + +func TestSetTranscriptStatus_RejectsInvalid(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + + id, err := s.UpsertVideo(ctx, ytVideo(userA, "bad12345", "bad status")) + require.NoError(t, err) + + require.Error(t, s.SetTranscriptStatus(ctx, userA, id, "bogus")) + + // The rejected write left the status untouched. + got, err := s.GetTranscriptStatus(ctx, userA, id) + require.NoError(t, err) + require.Equal(t, "", got) +} + +func TestSetTranscriptStatus_NotFound(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + + require.ErrorIs(t, s.SetTranscriptStatus(ctx, userA, videoX, "fetched"), store.ErrNotFound) + + _, err := s.GetTranscriptStatus(ctx, userA, videoX) + require.ErrorIs(t, err, store.ErrNotFound) +} + +func TestRateLimitedVideoIDs_StampsAndClears(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + + id, err := s.UpsertVideo(ctx, ytVideo(userA, "rl12345", "rate limited")) + require.NoError(t, err) + + // Marking rate_limited stamps rate_limited_at, so the video appears. + require.NoError(t, s.SetTranscriptStatus(ctx, userA, id, "rate_limited")) + rl, err := s.RateLimitedVideoIDs(ctx, userA) + require.NoError(t, err) + require.Contains(t, rl, id) + require.False(t, rl[id].IsZero(), "rate_limited_at must be stamped") + + // Moving off rate_limited clears the timestamp, so it drops out. + require.NoError(t, s.SetTranscriptStatus(ctx, userA, id, "fetched")) + rl, err = s.RateLimitedVideoIDs(ctx, userA) + require.NoError(t, err) + require.NotContains(t, rl, id) +}