feat(store): summarization-mode methods + all-videos read
Add the store surface for summarization mode: - SetAutoSummarize / GetAutoSummarize — per-user auto/manual toggle; an absent user reads as manual (the safe default). - RequestSummarize — queue one video (summarize_requested=TRUE); ErrNotFound when the video is absent or not owned (RLS hides another user's row). - RequestedVideoIDs / ClearSummarizeRequested — the run-loop side: load the queued set per pass (mirrors SeenVideoIDs), clear after summarizing. - ListVideos / GetVideoRow — drive from the videos table LEFT JOIN summaries so discovered-but-unsummarized videos appear with empty summary fields. SummaryRow gains additive Summarized + SummarizeRequested fields; the summary-only reads are untouched. RLS proof: rls_test.go gains a cross-user "queue B's video" write asserting it touches zero rows (store-level scoping can't prove this — the test pool is a superuser that bypasses RLS, same caveat documented there). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -39,6 +39,16 @@ type SummaryRow struct {
|
||||
FallbackUsed bool
|
||||
CreatedAt time.Time
|
||||
Actions []string // current active actions for this video; nil when none
|
||||
|
||||
// Summarized reports whether a summary exists for this video. The summary-only
|
||||
// reads (ListSummaries/GetSummaryByVideo) always yield true; the all-videos
|
||||
// read (ListVideos) yields false for a discovered-but-unsummarized video, whose
|
||||
// Summary/Highlights/AIProvider fields are then empty.
|
||||
Summarized bool
|
||||
// SummarizeRequested reflects videos.summarize_requested: the manual queue flag
|
||||
// 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
|
||||
}
|
||||
|
||||
// selectSummary is the shared projection for both reads. videos is LEFT JOINed
|
||||
@@ -101,6 +111,156 @@ func (s *Store) ListSummaries(ctx context.Context, userID string, limit int) ([]
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// selectVideo is the all-videos projection: it drives from the videos table and
|
||||
// LEFT JOINs the (at most one) summary, so a discovered-but-unsummarized video
|
||||
// still appears with empty summary fields. The column order mirrors selectSummary
|
||||
// for the shared fields, then appends summarized + summarize_requested. created_at
|
||||
// falls back to the video's seen_at when there is no summary, so the read-side row
|
||||
// always carries a sortable timestamp.
|
||||
const selectVideo = `
|
||||
SELECT v.id,
|
||||
v.provider_video_id,
|
||||
COALESCE(v.title, ''),
|
||||
v.provider,
|
||||
COALESCE(v.url, ''),
|
||||
v.published_at,
|
||||
COALESCE(s.summary, ''),
|
||||
s.highlights,
|
||||
s.takeaways,
|
||||
COALESCE(s.ai_provider, ''),
|
||||
COALESCE(s.ai_model, ''),
|
||||
COALESCE(s.fallback_used, FALSE),
|
||||
COALESCE(s.created_at, v.seen_at),
|
||||
(s.id IS NOT NULL) AS summarized,
|
||||
v.summarize_requested
|
||||
FROM videos v
|
||||
LEFT JOIN summaries s ON s.video_id = v.id AND s.user_id = v.user_id`
|
||||
|
||||
// ListVideos returns ALL of the user's videos — summarized and not — most recent
|
||||
// first by seen_at, capped at limit (non-positive defaults to 50). Unsummarized
|
||||
// videos come back with Summarized=false and empty summary fields, so the list
|
||||
// view can render them with a "Summarize" affordance. Scoped by user_id.
|
||||
func (s *Store) ListVideos(ctx context.Context, userID string, limit int) ([]SummaryRow, error) {
|
||||
if limit <= 0 {
|
||||
limit = 50
|
||||
}
|
||||
var out []SummaryRow
|
||||
if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
|
||||
rows, err := tx.Query(ctx,
|
||||
selectVideo+`
|
||||
WHERE v.user_id = $1
|
||||
ORDER BY v.seen_at DESC
|
||||
LIMIT $2`,
|
||||
userID, limit)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: list videos: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
row, err := scanVideoRow(rows)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
out = append(out, row)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return fmt.Errorf("store: iterate videos: %w", err)
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := s.attachActions(ctx, userID, out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// GetVideoRow returns a single video row (summarized or not) for (userID,
|
||||
// videoID), used to re-render one card after queuing it. Returns ErrNotFound when
|
||||
// the user has no such video. Scoped by user_id.
|
||||
func (s *Store) GetVideoRow(ctx context.Context, userID, videoID string) (*SummaryRow, error) {
|
||||
var (
|
||||
row SummaryRow
|
||||
found bool
|
||||
)
|
||||
if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
|
||||
rows, err := tx.Query(ctx,
|
||||
selectVideo+`
|
||||
WHERE v.user_id = $1 AND v.id = $2`,
|
||||
userID, videoID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: get video: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
if !rows.Next() {
|
||||
if err := rows.Err(); err != nil {
|
||||
return fmt.Errorf("store: get video: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
row, err = scanVideoRow(rows)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
found = true
|
||||
return nil
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !found {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
holder := []SummaryRow{row}
|
||||
if err := s.attachActions(ctx, userID, holder); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &holder[0], nil
|
||||
}
|
||||
|
||||
// scanVideoRow reads one row in the selectVideo column order. published_at is
|
||||
// nullable so it scans through a pointer.
|
||||
func scanVideoRow(rows pgx.Row) (SummaryRow, error) {
|
||||
var (
|
||||
row SummaryRow
|
||||
highlights []byte
|
||||
takeaways []byte
|
||||
publishedAt *time.Time
|
||||
)
|
||||
if err := rows.Scan(
|
||||
&row.VideoID,
|
||||
&row.ProviderVideoID,
|
||||
&row.Title,
|
||||
&row.Channel,
|
||||
&row.URL,
|
||||
&publishedAt,
|
||||
&row.Summary,
|
||||
&highlights,
|
||||
&takeaways,
|
||||
&row.AIProvider,
|
||||
&row.AIModel,
|
||||
&row.FallbackUsed,
|
||||
&row.CreatedAt,
|
||||
&row.Summarized,
|
||||
&row.SummarizeRequested,
|
||||
); err != nil {
|
||||
return SummaryRow{}, fmt.Errorf("store: scan video: %w", err)
|
||||
}
|
||||
if publishedAt != nil {
|
||||
row.PublishedAt = *publishedAt
|
||||
}
|
||||
var err error
|
||||
if row.Highlights, err = unmarshalList(highlights); err != nil {
|
||||
return SummaryRow{}, fmt.Errorf("store: unmarshal highlights: %w", err)
|
||||
}
|
||||
if row.Takeaways, err = unmarshalList(takeaways); err != nil {
|
||||
return SummaryRow{}, fmt.Errorf("store: unmarshal takeaways: %w", err)
|
||||
}
|
||||
return row, nil
|
||||
}
|
||||
|
||||
// GetSummaryByVideo returns the full summary for (userID, videoID), including
|
||||
// highlights and takeaways. Returns ErrNotFound when the user has no such
|
||||
// summary. Scoped by user_id.
|
||||
|
||||
@@ -181,6 +181,7 @@ func TestRLSEnforcesPerUserIsolation(t *testing.T) {
|
||||
}{
|
||||
{"update users", `UPDATE users SET display_name = 'hacked' WHERE id = $1`, b.userID},
|
||||
{"update videos", `UPDATE videos SET title = 'hacked' WHERE user_id = $1`, b.userID},
|
||||
{"queue videos summarize", `UPDATE videos SET summarize_requested = TRUE WHERE id = $1`, b.videoID},
|
||||
{"update transcripts", `UPDATE transcripts SET content = 'hacked' WHERE user_id = $1`, b.userID},
|
||||
{"update summaries", `UPDATE summaries SET summary = 'hacked' WHERE user_id = $1`, b.userID},
|
||||
{"update summary_actions", `UPDATE summary_actions SET action = 'skipped' WHERE user_id = $1`, b.userID},
|
||||
@@ -218,5 +219,10 @@ func TestRLSEnforcesPerUserIsolation(t *testing.T) {
|
||||
require.Equal(t, 1, bDeliveries, "A's DELETE must not have removed B's delivery")
|
||||
require.Equal(t, 1, bConnections, "A's writes must not have touched B's connection")
|
||||
|
||||
var bRequested bool
|
||||
require.NoError(t, super.QueryRow(ctx,
|
||||
`SELECT summarize_requested FROM videos WHERE user_id = $1`, b.userID).Scan(&bRequested))
|
||||
require.False(t, bRequested, "A scoped must not have queued B's video for summarization")
|
||||
|
||||
_ = a // a's ids are seeded for the symmetric read assertions above
|
||||
}
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// SetAutoSummarize sets the user's auto/manual summarization mode. TRUE =
|
||||
// automatic (every new video is summarized by `tapir run`); FALSE = manual (the
|
||||
// user queues videos individually). Per-user, not global (ADR-012). Scoped via
|
||||
// withUser, so RLS confines the UPDATE to the calling user's own row.
|
||||
func (s *Store) SetAutoSummarize(ctx context.Context, userID string, enabled bool) error {
|
||||
return s.withUser(ctx, userID, func(tx pgx.Tx) error {
|
||||
// Ensure the row exists (FK/identity target) before the UPDATE — mirrors
|
||||
// the Deliver/UpsertVideo paths, so toggling mode works even before the
|
||||
// first summary lands.
|
||||
if _, err := tx.Exec(ctx,
|
||||
`INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`,
|
||||
userID); err != nil {
|
||||
return fmt.Errorf("store: upsert user: %w", err)
|
||||
}
|
||||
if _, err := tx.Exec(ctx,
|
||||
`UPDATE users SET auto_summarize = $1 WHERE id = $2`, enabled, userID); err != nil {
|
||||
return fmt.Errorf("store: set auto summarize: %w", err)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// GetAutoSummarize reports the user's summarization mode (TRUE = automatic). An
|
||||
// absent user row reads as FALSE (manual), the safe default. Scoped via withUser.
|
||||
func (s *Store) GetAutoSummarize(ctx context.Context, userID string) (bool, error) {
|
||||
var enabled bool
|
||||
if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
|
||||
err := tx.QueryRow(ctx,
|
||||
`SELECT auto_summarize FROM users WHERE id = $1`, userID).Scan(&enabled)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
enabled = false
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}); err != nil {
|
||||
return false, fmt.Errorf("store: get auto summarize: %w", err)
|
||||
}
|
||||
return enabled, nil
|
||||
}
|
||||
|
||||
// RequestSummarize queues a single video for manual summarization by setting its
|
||||
// summarize_requested flag. The next `tapir run` picks it up and clears the flag.
|
||||
// Returns ErrNotFound when the video does not exist or is not owned by the user
|
||||
// (RLS hides another user's row, so the UPDATE matches zero rows). Scoped via
|
||||
// withUser.
|
||||
func (s *Store) RequestSummarize(ctx context.Context, userID, videoID string) error {
|
||||
return s.withUser(ctx, userID, func(tx pgx.Tx) error {
|
||||
ct, err := tx.Exec(ctx,
|
||||
`UPDATE videos SET summarize_requested = TRUE WHERE id = $1`, videoID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: request summarize: %w", err)
|
||||
}
|
||||
if ct.RowsAffected() == 0 {
|
||||
return ErrNotFound
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// RequestedVideoIDs returns the set of the user's video ids currently flagged for
|
||||
// manual summarization. The run loop loads it once per pass (mirroring
|
||||
// SeenVideoIDs) to decide which discovered videos to process in manual mode.
|
||||
// Scoped by user_id.
|
||||
func (s *Store) RequestedVideoIDs(ctx context.Context, userID string) (map[string]bool, error) {
|
||||
requested := make(map[string]bool)
|
||||
if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
|
||||
rows, err := tx.Query(ctx,
|
||||
`SELECT id FROM videos WHERE user_id = $1 AND summarize_requested = TRUE`, userID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: requested video ids: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
var id string
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
return fmt.Errorf("store: scan requested id: %w", err)
|
||||
}
|
||||
requested[id] = true
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return fmt.Errorf("store: iterate requested ids: %w", err)
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return requested, nil
|
||||
}
|
||||
|
||||
// ClearSummarizeRequested resets a video's manual queue flag, called by the run
|
||||
// loop after a queued video is successfully summarized so it is not re-processed
|
||||
// and the list view drops the "Queued" chip. Scoped via withUser.
|
||||
func (s *Store) ClearSummarizeRequested(ctx context.Context, userID, videoID string) error {
|
||||
return s.withUser(ctx, userID, func(tx pgx.Tx) error {
|
||||
if _, err := tx.Exec(ctx,
|
||||
`UPDATE videos SET summarize_requested = FALSE WHERE id = $1`, videoID); err != nil {
|
||||
return fmt.Errorf("store: clear summarize requested: %w", err)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,126 @@
|
||||
package store_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"gitea.d-ma.be/mathias/tapir/internal/adapters/store"
|
||||
)
|
||||
|
||||
// seedBareVideo inserts a videos row with no summary, so the all-videos read and
|
||||
// the manual-queue flag can be exercised without a delivered summary.
|
||||
func seedBareVideo(t *testing.T, p *pgxpool.Pool, userID, videoID, title string) {
|
||||
t.Helper()
|
||||
_, err := p.Exec(context.Background(),
|
||||
`INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, userID)
|
||||
require.NoError(t, err)
|
||||
_, err = p.Exec(context.Background(),
|
||||
`INSERT INTO videos (id, user_id, provider, provider_video_id, title)
|
||||
VALUES ($1, $2, 'youtube', $3, $4)`,
|
||||
videoID, userID, "pv-"+videoID[:8], title)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestAutoSummarizeRoundTripDefaultsFalse(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newStore(t)
|
||||
resetDB(t, rawPool(t))
|
||||
|
||||
// Unknown / fresh user defaults to manual (false).
|
||||
got, err := s.GetAutoSummarize(ctx, userA)
|
||||
require.NoError(t, err)
|
||||
require.False(t, got, "default mode is manual")
|
||||
|
||||
require.NoError(t, s.SetAutoSummarize(ctx, userA, true))
|
||||
got, err = s.GetAutoSummarize(ctx, userA)
|
||||
require.NoError(t, err)
|
||||
require.True(t, got, "set to automatic round-trips")
|
||||
|
||||
require.NoError(t, s.SetAutoSummarize(ctx, userA, false))
|
||||
got, err = s.GetAutoSummarize(ctx, userA)
|
||||
require.NoError(t, err)
|
||||
require.False(t, got, "set back to manual round-trips")
|
||||
}
|
||||
|
||||
func TestRequestSummarizeSetsFlag(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newStore(t)
|
||||
p := rawPool(t)
|
||||
resetDB(t, p)
|
||||
seedBareVideo(t, p, userA, videoX, "X Title")
|
||||
|
||||
require.NoError(t, s.RequestSummarize(ctx, userA, videoX))
|
||||
|
||||
requested, err := s.RequestedVideoIDs(ctx, userA)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, map[string]bool{videoX: true}, requested)
|
||||
|
||||
// Clearing drops it from the requested set.
|
||||
require.NoError(t, s.ClearSummarizeRequested(ctx, userA, videoX))
|
||||
requested, err = s.RequestedVideoIDs(ctx, userA)
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, requested)
|
||||
}
|
||||
|
||||
func TestRequestSummarizeMissingVideo(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newStore(t)
|
||||
resetDB(t, rawPool(t))
|
||||
|
||||
err := s.RequestSummarize(ctx, userA, videoX)
|
||||
require.ErrorIs(t, err, store.ErrNotFound, "queuing a non-existent video reports not found")
|
||||
}
|
||||
|
||||
func TestListVideosReturnsSummarizedAndUnsummarized(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newStore(t)
|
||||
p := rawPool(t)
|
||||
resetDB(t, p)
|
||||
|
||||
// videoX: discovered AND summarized. videoY: discovered, not yet summarized.
|
||||
seedBareVideo(t, p, userA, videoX, "Summarized One")
|
||||
seedBareVideo(t, p, userA, videoY, "Pending One")
|
||||
require.NoError(t, s.Deliver(ctx, summary(userA, videoX, "body x")))
|
||||
require.NoError(t, s.RequestSummarize(ctx, userA, videoY))
|
||||
|
||||
rows, err := s.ListVideos(ctx, userA, 50)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, rows, 2, "both summarized and unsummarized videos are listed")
|
||||
|
||||
byID := map[string]store.SummaryRow{}
|
||||
for _, r := range rows {
|
||||
byID[r.VideoID] = r
|
||||
}
|
||||
|
||||
require.True(t, byID[videoX].Summarized)
|
||||
require.Equal(t, "body x", byID[videoX].Summary)
|
||||
require.False(t, byID[videoX].SummarizeRequested)
|
||||
|
||||
require.False(t, byID[videoY].Summarized, "no summary -> Summarized false")
|
||||
require.Empty(t, byID[videoY].Summary, "unsummarized row has empty summary")
|
||||
require.True(t, byID[videoY].SummarizeRequested, "queued video carries the flag")
|
||||
}
|
||||
|
||||
func TestListVideosIsUserScoped(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newStore(t)
|
||||
p := rawPool(t)
|
||||
resetDB(t, p)
|
||||
seedBareVideo(t, p, userA, videoX, "A only")
|
||||
|
||||
rows, err := s.ListVideos(ctx, userB, 50)
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, rows, "user B must not see user A's videos")
|
||||
}
|
||||
|
||||
func TestGetVideoRowNotFound(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newStore(t)
|
||||
resetDB(t, rawPool(t))
|
||||
|
||||
_, err := s.GetVideoRow(ctx, userA, videoX)
|
||||
require.ErrorIs(t, err, store.ErrNotFound)
|
||||
}
|
||||
Reference in New Issue
Block a user