From 7b139c2cd73dc5a9fe7d4d9cded0615f04c0f2ae Mon Sep 17 00:00:00 2001 From: Mathias Date: Wed, 3 Jun 2026 15:15:52 +0200 Subject: [PATCH] feat(store): route all DB access through withUser for structural scoping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add Store.withUser(ctx, userID, fn) — a single choke point that BEGINs a tx, sets the transaction-local GUC tapir.current_user_id via set_config(..., true), runs fn, and commits. set_config is used over SET LOCAL because it is parameterizable; the local flag means the value auto-resets on commit/rollback so a pooled connection never leaks one request's user into the next. Route all 9 DB-touching methods through it (Deliver, HasSummary, SeenVideoIDs, ListSummaries, GetSummaryByVideo, SetAction, ClearAction, ActionsFor, and UpsertVideo; attachActions flows via ActionsFor). Scoping is now structural — not a per-query opt-in someone can forget — and arms the migration-003 RLS policies. Method signatures and existing WHERE clauses are unchanged (defence in depth; superuser DSNs in existing tests bypass RLS so behaviour is preserved). Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/adapters/store/actions.go | 101 ++++++++--------- internal/adapters/store/reads.go | 86 +++++++++------ internal/adapters/store/store.go | 168 ++++++++++++++++++----------- internal/adapters/store/videos.go | 51 +++++---- 4 files changed, 231 insertions(+), 175 deletions(-) diff --git a/internal/adapters/store/actions.go b/internal/adapters/store/actions.go index ead5e5d..fc713a6 100644 --- a/internal/adapters/store/actions.go +++ b/internal/adapters/store/actions.go @@ -3,6 +3,8 @@ package store import ( "context" "fmt" + + "github.com/jackc/pgx/v5" ) // allowedActions is the closed set of action verbs persisted in summary_actions. @@ -39,33 +41,25 @@ func (s *Store) SetAction(ctx context.Context, userID, videoID, action string) e return err } - tx, err := s.pool.Begin(ctx) - if err != nil { - return fmt.Errorf("store: begin set action: %w", err) - } - defer tx.Rollback(ctx) //nolint:errcheck // no-op after Commit - - if opposite, ok := oppositeAction[action]; ok { - if _, err := tx.Exec(ctx, - `DELETE FROM summary_actions - WHERE user_id = $1 AND video_id = $2 AND action = $3`, - userID, videoID, opposite); err != nil { - return fmt.Errorf("store: clear opposite action: %w", err) + return s.withUser(ctx, userID, func(tx pgx.Tx) error { + if opposite, ok := oppositeAction[action]; ok { + if _, err := tx.Exec(ctx, + `DELETE FROM summary_actions + WHERE user_id = $1 AND video_id = $2 AND action = $3`, + userID, videoID, opposite); err != nil { + return fmt.Errorf("store: clear opposite action: %w", err) + } } - } - if _, err := tx.Exec(ctx, - `INSERT INTO summary_actions (user_id, video_id, action) - VALUES ($1, $2, $3) - ON CONFLICT (user_id, video_id, action) DO UPDATE SET acted_at = NOW()`, - userID, videoID, action); err != nil { - return fmt.Errorf("store: set action: %w", err) - } - - if err := tx.Commit(ctx); err != nil { - return fmt.Errorf("store: commit set action: %w", err) - } - return nil + if _, err := tx.Exec(ctx, + `INSERT INTO summary_actions (user_id, video_id, action) + VALUES ($1, $2, $3) + ON CONFLICT (user_id, video_id, action) DO UPDATE SET acted_at = NOW()`, + userID, videoID, action); err != nil { + return fmt.Errorf("store: set action: %w", err) + } + return nil + }) } // ClearAction removes an action for (user, video). Clearing an action that is @@ -74,13 +68,15 @@ func (s *Store) ClearAction(ctx context.Context, userID, videoID, action string) if err := validateAction(action); err != nil { return err } - if _, err := s.pool.Exec(ctx, - `DELETE FROM summary_actions - WHERE user_id = $1 AND video_id = $2 AND action = $3`, - userID, videoID, action); err != nil { - return fmt.Errorf("store: clear action: %w", err) - } - return nil + return s.withUser(ctx, userID, func(tx pgx.Tx) error { + if _, err := tx.Exec(ctx, + `DELETE FROM summary_actions + WHERE user_id = $1 AND video_id = $2 AND action = $3`, + userID, videoID, action); err != nil { + return fmt.Errorf("store: clear action: %w", err) + } + return nil + }) } // ActionsFor returns the active actions per video for the given user, keyed by @@ -91,25 +87,30 @@ func (s *Store) ActionsFor(ctx context.Context, userID string, videoIDs []string if len(videoIDs) == 0 { return out, nil } - rows, err := s.pool.Query(ctx, - `SELECT video_id, action FROM summary_actions - WHERE user_id = $1 AND video_id = ANY($2) - ORDER BY video_id, action`, - userID, videoIDs) - if err != nil { - return nil, fmt.Errorf("store: actions for: %w", err) - } - defer rows.Close() - - for rows.Next() { - var videoID, action string - if err := rows.Scan(&videoID, &action); err != nil { - return nil, fmt.Errorf("store: scan action: %w", err) + if err := s.withUser(ctx, userID, func(tx pgx.Tx) error { + rows, err := tx.Query(ctx, + `SELECT video_id, action FROM summary_actions + WHERE user_id = $1 AND video_id = ANY($2) + ORDER BY video_id, action`, + userID, videoIDs) + if err != nil { + return fmt.Errorf("store: actions for: %w", err) } - out[videoID] = append(out[videoID], action) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate actions: %w", err) + defer rows.Close() + + for rows.Next() { + var videoID, action string + if err := rows.Scan(&videoID, &action); err != nil { + return fmt.Errorf("store: scan action: %w", err) + } + out[videoID] = append(out[videoID], action) + } + if err := rows.Err(); err != nil { + return fmt.Errorf("store: iterate actions: %w", err) + } + return nil + }); err != nil { + return nil, err } return out, nil } diff --git a/internal/adapters/store/reads.go b/internal/adapters/store/reads.go index 8233d1a..78c294c 100644 --- a/internal/adapters/store/reads.go +++ b/internal/adapters/store/reads.go @@ -66,27 +66,32 @@ func (s *Store) ListSummaries(ctx context.Context, userID string, limit int) ([] if limit <= 0 { limit = 50 } - rows, err := s.pool.Query(ctx, - selectSummary+` - WHERE s.user_id = $1 - ORDER BY s.created_at DESC - LIMIT $2`, - userID, limit) - if err != nil { - return nil, fmt.Errorf("store: list summaries: %w", err) - } - defer rows.Close() - var out []SummaryRow - for rows.Next() { - row, err := scanSummaryRow(rows) + if err := s.withUser(ctx, userID, func(tx pgx.Tx) error { + rows, err := tx.Query(ctx, + selectSummary+` + WHERE s.user_id = $1 + ORDER BY s.created_at DESC + LIMIT $2`, + userID, limit) if err != nil { - return nil, err + return fmt.Errorf("store: list summaries: %w", err) } - out = append(out, row) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate summaries: %w", err) + defer rows.Close() + + for rows.Next() { + row, err := scanSummaryRow(rows) + if err != nil { + return err + } + out = append(out, row) + } + if err := rows.Err(); err != nil { + return fmt.Errorf("store: iterate summaries: %w", err) + } + return nil + }); err != nil { + return nil, err } if err := s.attachActions(ctx, userID, out); err != nil { return nil, err @@ -98,25 +103,38 @@ func (s *Store) ListSummaries(ctx context.Context, userID string, limit int) ([] // highlights and takeaways. Returns ErrNotFound when the user has no such // summary. Scoped by user_id. func (s *Store) GetSummaryByVideo(ctx context.Context, userID, videoID string) (*SummaryRow, error) { - rows, err := s.pool.Query(ctx, - selectSummary+` - WHERE s.user_id = $1 AND s.video_id = $2`, - userID, videoID) - if err != nil { - return nil, fmt.Errorf("store: get summary: %w", err) - } - defer rows.Close() - - if !rows.Next() { - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: get summary: %w", err) + var ( + row SummaryRow + found bool + ) + if err := s.withUser(ctx, userID, func(tx pgx.Tx) error { + rows, err := tx.Query(ctx, + selectSummary+` + WHERE s.user_id = $1 AND s.video_id = $2`, + userID, videoID) + if err != nil { + return fmt.Errorf("store: get summary: %w", err) } - return nil, ErrNotFound - } - row, err := scanSummaryRow(rows) - if err != nil { + defer rows.Close() + + if !rows.Next() { + if err := rows.Err(); err != nil { + return fmt.Errorf("store: get summary: %w", err) + } + return nil + } + row, err = scanSummaryRow(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 diff --git a/internal/adapters/store/store.go b/internal/adapters/store/store.go index 16737d4..724a3f5 100644 --- a/internal/adapters/store/store.go +++ b/internal/adapters/store/store.go @@ -19,6 +19,7 @@ import ( "github.com/golang-migrate/migrate/v4" migratepgx "github.com/golang-migrate/migrate/v4/database/pgx/v5" "github.com/golang-migrate/migrate/v4/source/iofs" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" _ "github.com/jackc/pgx/v5/stdlib" // register the "pgx" database/sql driver for migrate @@ -90,6 +91,46 @@ func (s *Store) Close() { // Name identifies this sink in delivery records. func (s *Store) Name() string { return "store" } +// withUser is the single choke point through which EVERY DB access in this +// package flows, so per-user isolation is structural — not a per-query opt-in +// someone can forget. It: +// +// - BEGINs a transaction, +// - sets the per-request GUC tapir.current_user_id via +// set_config('tapir.current_user_id', $1, true). The set_config form is used +// instead of `SET LOCAL` because it is parameterizable (SET cannot bind a +// value through the driver); the third arg true = local = transaction-scoped, +// so it auto-resets on commit/rollback and a pooled connection never leaks one +// request's user into the next, +// - runs fn against that transaction, +// - COMMITs (or ROLLBACKs on error). +// +// The migration-003 RLS policies key off this GUC: a row is visible/writable only +// when its owner = current_setting('tapir.current_user_id'). RLS enforces only +// when the app connects as a non-superuser, non-BYPASSRLS role (in production the +// table owner tapir, made subject via FORCE ROW LEVEL SECURITY). A superuser DSN +// bypasses RLS regardless — see rls_test.go, which connects as a dedicated +// non-superuser role to prove the enforcement is real. +func (s *Store) withUser(ctx context.Context, userID string, fn func(pgx.Tx) error) error { + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("store: begin: %w", err) + } + defer tx.Rollback(ctx) //nolint:errcheck // no-op after Commit + + if _, err := tx.Exec(ctx, + `SELECT set_config('tapir.current_user_id', $1, true)`, userID); err != nil { + return fmt.Errorf("store: scope user: %w", err) + } + if err := fn(tx); err != nil { + return err + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("store: commit: %w", err) + } + return nil +} + // Deliver upserts the summary idempotently on (user_id, video_id) and records the // store delivery. Re-delivering the same summary updates in place — it never // errors or duplicates. The whole write is one transaction so a summary and its @@ -104,63 +145,57 @@ func (s *Store) Deliver(ctx context.Context, sum domain.Summary) error { return fmt.Errorf("store: marshal takeaways: %w", err) } - tx, err := s.pool.Begin(ctx) - if err != nil { - return fmt.Errorf("store: begin: %w", err) - } - defer tx.Rollback(ctx) //nolint:errcheck // no-op after Commit + return s.withUser(ctx, sum.UserID, func(tx pgx.Tx) error { + // Ensure the owning user exists (FK target). The store sink receives only + // a Summary, so a minimal user row is enough at Stage 0. + if _, err := tx.Exec(ctx, + `INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, + sum.UserID); err != nil { + return fmt.Errorf("store: upsert user: %w", err) + } - // Ensure the owning user exists (FK target). The store sink receives only a - // Summary, so a minimal user row is enough at Stage 0. - if _, err := tx.Exec(ctx, - `INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, - sum.UserID); err != nil { - return fmt.Errorf("store: upsert user: %w", err) - } + var summaryID string + if err := tx.QueryRow(ctx, + `INSERT INTO summaries + (user_id, video_id, summary, highlights, takeaways, ai_provider, ai_model, fallback_used) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + ON CONFLICT (user_id, video_id) DO UPDATE SET + summary = EXCLUDED.summary, + highlights = EXCLUDED.highlights, + takeaways = EXCLUDED.takeaways, + ai_provider = EXCLUDED.ai_provider, + ai_model = EXCLUDED.ai_model, + fallback_used = EXCLUDED.fallback_used + RETURNING id`, + sum.UserID, sum.VideoID, sum.Summary, highlights, takeaways, + sum.AIProvider, sum.AIModel, sum.FallbackUsed, + ).Scan(&summaryID); err != nil { + return fmt.Errorf("store: upsert summary: %w", err) + } - var summaryID string - if err := tx.QueryRow(ctx, - `INSERT INTO summaries - (user_id, video_id, summary, highlights, takeaways, ai_provider, ai_model, fallback_used) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8) - ON CONFLICT (user_id, video_id) DO UPDATE SET - summary = EXCLUDED.summary, - highlights = EXCLUDED.highlights, - takeaways = EXCLUDED.takeaways, - ai_provider = EXCLUDED.ai_provider, - ai_model = EXCLUDED.ai_model, - fallback_used = EXCLUDED.fallback_used - RETURNING id`, - sum.UserID, sum.VideoID, sum.Summary, highlights, takeaways, - sum.AIProvider, sum.AIModel, sum.FallbackUsed, - ).Scan(&summaryID); err != nil { - return fmt.Errorf("store: upsert summary: %w", err) - } - - if _, err := tx.Exec(ctx, - `INSERT INTO sink_deliveries (summary_id, sink, status) - VALUES ($1, 'store', 'delivered') - ON CONFLICT (summary_id, sink) DO UPDATE SET - status = 'delivered', - detail = NULL, - updated_at = NOW()`, - summaryID); err != nil { - return fmt.Errorf("store: record delivery: %w", err) - } - - if err := tx.Commit(ctx); err != nil { - return fmt.Errorf("store: commit: %w", err) - } - return nil + if _, err := tx.Exec(ctx, + `INSERT INTO sink_deliveries (summary_id, sink, status) + VALUES ($1, 'store', 'delivered') + ON CONFLICT (summary_id, sink) DO UPDATE SET + status = 'delivered', + detail = NULL, + updated_at = NOW()`, + summaryID); err != nil { + return fmt.Errorf("store: record delivery: %w", err) + } + return nil + }) } // HasSummary reports whether a summary already exists for (userID, videoID). // This is the per-video durable dedup check. func (s *Store) HasSummary(ctx context.Context, userID, videoID string) (bool, error) { var exists bool - if err := s.pool.QueryRow(ctx, - `SELECT EXISTS(SELECT 1 FROM summaries WHERE user_id = $1 AND video_id = $2)`, - userID, videoID).Scan(&exists); err != nil { + if err := s.withUser(ctx, userID, func(tx pgx.Tx) error { + return tx.QueryRow(ctx, + `SELECT EXISTS(SELECT 1 FROM summaries WHERE user_id = $1 AND video_id = $2)`, + userID, videoID).Scan(&exists) + }); err != nil { return false, fmt.Errorf("store: has summary: %w", err) } return exists, nil @@ -170,23 +205,28 @@ func (s *Store) HasSummary(ctx context.Context, userID, videoID string) (bool, e // user. The watcher uses it to skip re-summarizing across restarts. Scoped by // user_id, so one user never sees another's videos. func (s *Store) SeenVideoIDs(ctx context.Context, userID string) (map[string]bool, error) { - rows, err := s.pool.Query(ctx, - `SELECT video_id FROM summaries WHERE user_id = $1`, userID) - if err != nil { - return nil, fmt.Errorf("store: seen video ids: %w", err) - } - defer rows.Close() - seen := make(map[string]bool) - for rows.Next() { - var id string - if err := rows.Scan(&id); err != nil { - return nil, fmt.Errorf("store: scan video id: %w", err) + if err := s.withUser(ctx, userID, func(tx pgx.Tx) error { + rows, err := tx.Query(ctx, + `SELECT video_id FROM summaries WHERE user_id = $1`, userID) + if err != nil { + return fmt.Errorf("store: seen video ids: %w", err) } - seen[id] = true - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("store: iterate 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 video id: %w", err) + } + seen[id] = true + } + if err := rows.Err(); err != nil { + return fmt.Errorf("store: iterate video ids: %w", err) + } + return nil + }); err != nil { + return nil, err } return seen, nil } diff --git a/internal/adapters/store/videos.go b/internal/adapters/store/videos.go index 64e63a5..8a0e13a 100644 --- a/internal/adapters/store/videos.go +++ b/internal/adapters/store/videos.go @@ -5,6 +5,8 @@ import ( "fmt" "time" + "github.com/jackc/pgx/v5" + "gitea.d-ma.be/mathias/tapir/internal/domain" ) @@ -29,40 +31,35 @@ func (s *Store) UpsertVideo(ctx context.Context, v domain.Video) (string, error) return "", fmt.Errorf("store: upsert video: empty provider video id") } - tx, err := s.pool.Begin(ctx) - if err != nil { - return "", fmt.Errorf("store: begin: %w", err) - } - defer tx.Rollback(ctx) //nolint:errcheck // no-op after Commit - - // Ensure the owning user exists (FK target) — same as the Deliver path. - if _, err := tx.Exec(ctx, - `INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, - v.UserID); err != nil { - return "", fmt.Errorf("store: upsert user: %w", err) - } - provider := string(v.Provider) if provider == "" { provider = string(domain.ProviderYouTube) } var id string - if err := tx.QueryRow(ctx, - `INSERT INTO videos (user_id, provider, provider_video_id, title, url, published_at) - VALUES ($1, $2, $3, $4, $5, $6) - ON CONFLICT (user_id, provider, provider_video_id) DO UPDATE SET - title = EXCLUDED.title, - url = EXCLUDED.url, - published_at = EXCLUDED.published_at - RETURNING id`, - v.UserID, provider, v.ProviderVideoID, v.Title, v.URL, nullTime(v.PublishedAt), - ).Scan(&id); err != nil { - return "", fmt.Errorf("store: upsert video: %w", err) - } + if err := s.withUser(ctx, v.UserID, func(tx pgx.Tx) error { + // Ensure the owning user exists (FK target) — same as the Deliver path. + if _, err := tx.Exec(ctx, + `INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, + v.UserID); err != nil { + return fmt.Errorf("store: upsert user: %w", err) + } - if err := tx.Commit(ctx); err != nil { - return "", fmt.Errorf("store: commit: %w", err) + if err := tx.QueryRow(ctx, + `INSERT INTO videos (user_id, provider, provider_video_id, title, url, published_at) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (user_id, provider, provider_video_id) DO UPDATE SET + title = EXCLUDED.title, + url = EXCLUDED.url, + published_at = EXCLUDED.published_at + RETURNING id`, + v.UserID, provider, v.ProviderVideoID, v.Title, v.URL, nullTime(v.PublishedAt), + ).Scan(&id); err != nil { + return fmt.Errorf("store: upsert video: %w", err) + } + return nil + }); err != nil { + return "", err } return id, nil }