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/migrations/003_rls.down.sql b/internal/adapters/store/migrations/003_rls.down.sql new file mode 100644 index 0000000..87cdf0f --- /dev/null +++ b/internal/adapters/store/migrations/003_rls.down.sql @@ -0,0 +1,23 @@ +DROP POLICY IF EXISTS sink_deliveries_isolation ON sink_deliveries; +ALTER TABLE sink_deliveries NO FORCE ROW LEVEL SECURITY; +ALTER TABLE sink_deliveries DISABLE ROW LEVEL SECURITY; + +DROP POLICY IF EXISTS summary_actions_isolation ON summary_actions; +ALTER TABLE summary_actions NO FORCE ROW LEVEL SECURITY; +ALTER TABLE summary_actions DISABLE ROW LEVEL SECURITY; + +DROP POLICY IF EXISTS summaries_isolation ON summaries; +ALTER TABLE summaries NO FORCE ROW LEVEL SECURITY; +ALTER TABLE summaries DISABLE ROW LEVEL SECURITY; + +DROP POLICY IF EXISTS transcripts_isolation ON transcripts; +ALTER TABLE transcripts NO FORCE ROW LEVEL SECURITY; +ALTER TABLE transcripts DISABLE ROW LEVEL SECURITY; + +DROP POLICY IF EXISTS videos_isolation ON videos; +ALTER TABLE videos NO FORCE ROW LEVEL SECURITY; +ALTER TABLE videos DISABLE ROW LEVEL SECURITY; + +DROP POLICY IF EXISTS users_isolation ON users; +ALTER TABLE users NO FORCE ROW LEVEL SECURITY; +ALTER TABLE users DISABLE ROW LEVEL SECURITY; diff --git a/internal/adapters/store/migrations/003_rls.up.sql b/internal/adapters/store/migrations/003_rls.up.sql new file mode 100644 index 0000000..6135fc6 --- /dev/null +++ b/internal/adapters/store/migrations/003_rls.up.sql @@ -0,0 +1,68 @@ +-- Migration 003: enforce per-user isolation at the DB layer via row-level +-- security (ADR-012, data-model.md "Isolation invariant"). Stage 1 ships +-- multi-user WITH this enforcement; it is the proof that user A cannot read or +-- write user B's rows even if application-level WHERE clauses are wrong. +-- +-- How it works: +-- * Every policy keys off the per-request GUC tapir.current_user_id, set by the +-- store's withUser helper via set_config('tapir.current_user_id', $1, true) +-- (transaction-local — auto-reset on commit/rollback, never leaks across a +-- pooled connection's requests). +-- * current_setting('tapir.current_user_id', true) uses missing_ok = true: an +-- UNSET GUC yields NULL, so the predicate is NULL → no rows match → deny-all. +-- That is the safe default and is asserted in rls_test.go. +-- * FORCE ROW LEVEL SECURITY: the app connects as the table OWNER (tapir), and +-- owners BYPASS RLS unless forced. Without FORCE the policies below are dead +-- for the production user. FORCE makes the owner subject to them. (A superuser +-- DSN still bypasses RLS regardless — the test connects as a non-superuser, +-- non-BYPASSRLS role so the enforcement is real, not theatre.) + +-- users: the row's own id IS the user_id for this table. +ALTER TABLE users ENABLE ROW LEVEL SECURITY; +ALTER TABLE users FORCE ROW LEVEL SECURITY; +CREATE POLICY users_isolation ON users + FOR ALL + USING (id = current_setting('tapir.current_user_id', true)::uuid); + +ALTER TABLE videos ENABLE ROW LEVEL SECURITY; +ALTER TABLE videos FORCE ROW LEVEL SECURITY; +CREATE POLICY videos_isolation ON videos + FOR ALL + USING (user_id = current_setting('tapir.current_user_id', true)::uuid); + +ALTER TABLE transcripts ENABLE ROW LEVEL SECURITY; +ALTER TABLE transcripts FORCE ROW LEVEL SECURITY; +CREATE POLICY transcripts_isolation ON transcripts + FOR ALL + USING (user_id = current_setting('tapir.current_user_id', true)::uuid); + +ALTER TABLE summaries ENABLE ROW LEVEL SECURITY; +ALTER TABLE summaries FORCE ROW LEVEL SECURITY; +CREATE POLICY summaries_isolation ON summaries + FOR ALL + USING (user_id = current_setting('tapir.current_user_id', true)::uuid); + +ALTER TABLE summary_actions ENABLE ROW LEVEL SECURITY; +ALTER TABLE summary_actions FORCE ROW LEVEL SECURITY; +CREATE POLICY summary_actions_isolation ON summary_actions + FOR ALL + USING (user_id = current_setting('tapir.current_user_id', true)::uuid); + +-- sink_deliveries has NO user_id of its own; ownership is derived from the +-- summary it belongs to. We key the policy directly off the GUC via EXISTS +-- (rather than `summary_id IN (SELECT id FROM summaries)`) so it is self-contained +-- and does not silently depend on summaries' own RLS being applied to the +-- subquery. The WITH CHECK clause (defaulting to USING under FOR ALL) means a +-- delivery row can only be inserted/updated when its summary is owned by the +-- current user. +ALTER TABLE sink_deliveries ENABLE ROW LEVEL SECURITY; +ALTER TABLE sink_deliveries FORCE ROW LEVEL SECURITY; +CREATE POLICY sink_deliveries_isolation ON sink_deliveries + FOR ALL + USING ( + EXISTS ( + SELECT 1 FROM summaries s + WHERE s.id = sink_deliveries.summary_id + AND s.user_id = current_setting('tapir.current_user_id', true)::uuid + ) + ); 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/rls_test.go b/internal/adapters/store/rls_test.go new file mode 100644 index 0000000..7eb8e07 --- /dev/null +++ b/internal/adapters/store/rls_test.go @@ -0,0 +1,212 @@ +package store_test + +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/require" +) + +// This is the isolation proof for ADR-012: per-user isolation is enforced by the +// database (migration 003 RLS policies), not merely by application WHERE clauses. +// +// CRITICAL: embedded-postgres's default user (postgres) is a SUPERUSER, which +// BYPASSES RLS regardless of FORCE ROW LEVEL SECURITY. A test that ran scoped +// queries as postgres would be fake-green — it would pass even with the policies +// removed. So this test creates a dedicated NON-SUPERUSER, non-BYPASSRLS role +// ("app", mirroring the production table-owner role tapir which FORCE subjects to +// RLS) and runs every scoped query as that role. The deny-all sanity check below +// (no GUC set → zero rows) proves the enforcement path is live, not bypassed. + +// userIsolatedTables are the tables that carry a user_id and whose policy keys +// directly off the tapir.current_user_id GUC. +var userIsolatedTables = []string{ + "users", "videos", "transcripts", "summaries", "summary_actions", +} + +// allIsolatedTables adds sink_deliveries, whose ownership is derived from its +// summary (no user_id column of its own). +var allIsolatedTables = append(append([]string{}, userIsolatedTables...), "sink_deliveries") + +// seeded captures the DB-generated ids for one user's row chain. +type seeded struct { + userID string + videoID string // videos.id (UUID), reused as summaries.video_id + summaryID string +} + +// seedUser inserts one full chain (user → video → transcript → summary → +// action → delivery) as the superuser pool, which bypasses RLS so both users' +// data lands regardless of the GUC. +func seedUser(t *testing.T, p *pgxpool.Pool, userID string) seeded { + t.Helper() + ctx := context.Background() + + _, err := p.Exec(ctx, `INSERT INTO users (id) VALUES ($1)`, userID) + require.NoError(t, err) + + var videoID string + require.NoError(t, p.QueryRow(ctx, + `INSERT INTO videos (user_id, provider, provider_video_id, title) + VALUES ($1, 'youtube', $2, 'title') RETURNING id`, + userID, "vid-"+userID).Scan(&videoID)) + + _, err = p.Exec(ctx, + `INSERT INTO transcripts (video_id, user_id, source, content) + VALUES ($1, $2, 'captions', 'words')`, videoID, userID) + require.NoError(t, err) + + var summaryID string + require.NoError(t, p.QueryRow(ctx, + `INSERT INTO summaries (user_id, video_id, summary) VALUES ($1, $2, 'sum') + RETURNING id`, userID, videoID).Scan(&summaryID)) + + _, err = p.Exec(ctx, + `INSERT INTO summary_actions (user_id, video_id, action) + VALUES ($1, $2, 'watched')`, userID, videoID) + require.NoError(t, err) + + _, err = p.Exec(ctx, + `INSERT INTO sink_deliveries (summary_id, sink, status) + VALUES ($1, 'store', 'delivered')`, summaryID) + require.NoError(t, err) + + return seeded{userID: userID, videoID: videoID, summaryID: summaryID} +} + +// appPool creates a non-superuser role with DML grants and returns a pool +// connected AS that role, so RLS is actually enforced for it. +func appPool(t *testing.T, super *pgxpool.Pool) *pgxpool.Pool { + t.Helper() + ctx := context.Background() + + // Idempotent across test runs (schema/role persist for the TestMain PG). + _, _ = super.Exec(ctx, `DROP ROLE IF EXISTS app`) + _, err := super.Exec(ctx, `CREATE ROLE app LOGIN PASSWORD 'app'`) + require.NoError(t, err) + _, err = super.Exec(ctx, `GRANT USAGE ON SCHEMA public TO app`) + require.NoError(t, err) + _, err = super.Exec(ctx, + `GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO app`) + require.NoError(t, err) + + appDSN := strings.Replace(dsn, "postgres:postgres@", "app:app@", 1) + p, err := pgxpool.New(ctx, appDSN) + require.NoError(t, err) + t.Cleanup(p.Close) + + // Sanity: the app role must NOT be a superuser / must not bypass RLS, else + // this whole test is theatre. + var isSuper bool + require.NoError(t, p.QueryRow(ctx, + `SELECT rolsuper FROM pg_roles WHERE rolname = current_user`).Scan(&isSuper)) + require.False(t, isSuper, "app role must be non-superuser or RLS is bypassed") + return p +} + +// scopedCount counts rows in table as the app role, optionally scoped to a user +// via the transaction-local GUC. An empty scope sets no GUC (deny-all path). +func scopedCount(t *testing.T, p *pgxpool.Pool, scope, table string) int { + t.Helper() + ctx := context.Background() + tx, err := p.Begin(ctx) + require.NoError(t, err) + defer tx.Rollback(ctx) //nolint:errcheck + + if scope != "" { + _, err = tx.Exec(ctx, `SELECT set_config('tapir.current_user_id', $1, true)`, scope) + require.NoError(t, err) + } + var n int + require.NoError(t, tx.QueryRow(ctx, `SELECT count(*) FROM `+table).Scan(&n)) + return n +} + +// scopedRowsAffected runs a write as the app role scoped to scope and returns the +// rows affected, so we can assert a cross-user write touches zero rows. +func scopedRowsAffected(t *testing.T, p *pgxpool.Pool, scope, sql string, args ...any) int64 { + t.Helper() + ctx := context.Background() + tx, err := p.Begin(ctx) + require.NoError(t, err) + defer tx.Rollback(ctx) //nolint:errcheck + + _, err = tx.Exec(ctx, `SELECT set_config('tapir.current_user_id', $1, true)`, scope) + require.NoError(t, err) + ct, err := tx.Exec(ctx, sql, args...) + require.NoError(t, err) // RLS hides the rows; it is NOT a permission error + require.NoError(t, tx.Commit(ctx)) + return ct.RowsAffected() +} + +func TestRLSEnforcesPerUserIsolation(t *testing.T) { + newStore(t) // apply migrations (incl. 003 RLS) as superuser + super := rawPool(t) + resetDB(t, super) + + a := seedUser(t, super, userA) + b := seedUser(t, super, userB) + app := appPool(t, super) + + // 1. Deny-all: with NO GUC set, every isolated table returns zero rows. This + // proves RLS is actually ON (a bypassed/superuser path would see all rows). + for _, table := range allIsolatedTables { + require.Equal(t, 0, scopedCount(t, app, "", table), + "unset tapir.current_user_id must yield deny-all on %s", table) + } + + // 2. Scoped reads: A sees exactly its own one row per table; likewise B. A + // seeing B's row (or vice versa) would mean isolation is broken. + for _, table := range allIsolatedTables { + require.Equal(t, 1, scopedCount(t, app, userA, table), + "user A scoped read must see exactly its own row in %s", table) + require.Equal(t, 1, scopedCount(t, app, userB, table), + "user B scoped read must see exactly its own row in %s", table) + } + + // 3. Cross-user writes are invisible: scoped to A, an UPDATE/DELETE aimed at + // B's rows affects zero rows (RLS hides them from the write, too). + writes := []struct { + name string + sql string + arg any // identifies B's row(s) + }{ + {"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}, + {"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}, + {"update sink_deliveries", `UPDATE sink_deliveries SET status = 'hacked' WHERE summary_id = $1`, b.summaryID}, + {"delete summaries", `DELETE FROM summaries WHERE user_id = $1`, b.userID}, + {"delete summary_actions", `DELETE FROM summary_actions WHERE user_id = $1`, b.userID}, + {"delete sink_deliveries", `DELETE FROM sink_deliveries WHERE summary_id = $1`, b.summaryID}, + } + for _, w := range writes { + require.Equal(t, int64(0), scopedRowsAffected(t, app, userA, w.sql, w.arg), + "user A scoped %s must touch zero of user B's rows", w.name) + } + + // 4. B's rows survived unchanged (the writes above neither modified nor + // deleted them), verified via the superuser pool which bypasses RLS. + ctx := context.Background() + var bSummary string + require.NoError(t, super.QueryRow(ctx, + `SELECT summary FROM summaries WHERE user_id = $1`, b.userID).Scan(&bSummary)) + require.Equal(t, "sum", bSummary, "B's summary must be untouched by A's writes") + + var bSummaries, bActions, bDeliveries int + require.NoError(t, super.QueryRow(ctx, + `SELECT count(*) FROM summaries WHERE user_id = $1`, b.userID).Scan(&bSummaries)) + require.NoError(t, super.QueryRow(ctx, + `SELECT count(*) FROM summary_actions WHERE user_id = $1`, b.userID).Scan(&bActions)) + require.NoError(t, super.QueryRow(ctx, + fmt.Sprintf(`SELECT count(*) FROM sink_deliveries WHERE summary_id = '%s'`, b.summaryID)).Scan(&bDeliveries)) + require.Equal(t, 1, bSummaries, "A's DELETE must not have removed B's summary") + require.Equal(t, 1, bActions, "A's DELETE must not have removed B's action") + require.Equal(t, 1, bDeliveries, "A's DELETE must not have removed B's delivery") + + _ = a // a's ids are seeded for the symmetric read assertions above +} 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 }