From bdbdce7de187a9c0c140feb8ba62283e03b3f3d8 Mon Sep 17 00:00:00 2001 From: Mathias Date: Wed, 3 Jun 2026 19:26:52 +0200 Subject: [PATCH] feat(store): summarization-mode methods + all-videos read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- internal/adapters/store/reads.go | 160 ++++++++++++++++++ internal/adapters/store/rls_test.go | 6 + internal/adapters/store/summarize_mode.go | 112 ++++++++++++ .../adapters/store/summarize_mode_test.go | 126 ++++++++++++++ 4 files changed, 404 insertions(+) create mode 100644 internal/adapters/store/summarize_mode.go create mode 100644 internal/adapters/store/summarize_mode_test.go diff --git a/internal/adapters/store/reads.go b/internal/adapters/store/reads.go index 4ca1336..bd0b742 100644 --- a/internal/adapters/store/reads.go +++ b/internal/adapters/store/reads.go @@ -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. diff --git a/internal/adapters/store/rls_test.go b/internal/adapters/store/rls_test.go index b01176d..b4b7385 100644 --- a/internal/adapters/store/rls_test.go +++ b/internal/adapters/store/rls_test.go @@ -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 } diff --git a/internal/adapters/store/summarize_mode.go b/internal/adapters/store/summarize_mode.go new file mode 100644 index 0000000..67bd54c --- /dev/null +++ b/internal/adapters/store/summarize_mode.go @@ -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 + }) +} diff --git a/internal/adapters/store/summarize_mode_test.go b/internal/adapters/store/summarize_mode_test.go new file mode 100644 index 0000000..30d60dc --- /dev/null +++ b/internal/adapters/store/summarize_mode_test.go @@ -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) +}