feat(store): TranscriptStatus read field + status setter/loader
Surfaces videos.transcript_status (migration 007) on SummaryRow and adds SetTranscriptStatus / GetTranscriptStatus / RateLimitedVideoIDs. SetTranscriptStatus is the single choke point for the rate-limit lifecycle: "rate_limited" stamps rate_limited_at = NOW(), every other status clears it, so the runner's backoff window and the UI badge read one consistent source. RateLimitedVideoIDs is the per-pass loader (mirrors SeenVideoIDs) the runner uses to skip still-throttled videos without re-hitting the caption endpoint. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -49,6 +49,10 @@ type SummaryRow struct {
|
|||||||
// set by the web "Summarize" button and cleared by the next `tapir run`. Only
|
// set by the web "Summarize" button and cleared by the next `tapir run`. Only
|
||||||
// populated by ListVideos/GetVideoRow (summary-only reads leave it false).
|
// populated by ListVideos/GetVideoRow (summary-only reads leave it false).
|
||||||
SummarizeRequested bool
|
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
|
// 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.fallback_used, FALSE),
|
||||||
COALESCE(s.created_at, v.seen_at),
|
COALESCE(s.created_at, v.seen_at),
|
||||||
(s.id IS NOT NULL) AS summarized,
|
(s.id IS NOT NULL) AS summarized,
|
||||||
v.summarize_requested
|
v.summarize_requested,
|
||||||
|
COALESCE(v.transcript_status, '')
|
||||||
FROM videos v
|
FROM videos v
|
||||||
LEFT JOIN summaries s ON s.video_id = v.id AND s.user_id = v.user_id`
|
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.CreatedAt,
|
||||||
&row.Summarized,
|
&row.Summarized,
|
||||||
&row.SummarizeRequested,
|
&row.SummarizeRequested,
|
||||||
|
&row.TranscriptStatus,
|
||||||
); err != nil {
|
); err != nil {
|
||||||
return SummaryRow{}, fmt.Errorf("store: scan video: %w", err)
|
return SummaryRow{}, fmt.Errorf("store: scan video: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user