merge: RLS isolation foundation — forced RLS + withUser scoping + isolation test (Worker I, ADR-012)

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,12 +41,7 @@ func (s *Store) SetAction(ctx context.Context, userID, videoID, action string) e
 		return err
 	}
 
-	tx, err := s.pool.Begin(ctx)
+	return s.withUser(ctx, userID, func(tx pgx.Tx) error {
-	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
@@ -61,11 +58,8 @@ func (s *Store) SetAction(ctx context.Context, userID, videoID, action string) e
 			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
+	})
 }
 
 // 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,
+	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,
+	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 nil, fmt.Errorf("store: actions for: %w", err)
+			return 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)
+				return fmt.Errorf("store: scan action: %w", err)
 			}
 			out[videoID] = append(out[videoID], action)
 		}
 		if err := rows.Err(); err != nil {
-		return nil, fmt.Errorf("store: iterate actions: %w", err)
+			return fmt.Errorf("store: iterate actions: %w", err)
+		}
+		return nil
+	}); err != nil {
+		return nil, err
 	}
 	return out, nil
 }

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;

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
+        )
+    );

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,
+	var out []SummaryRow
+	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, fmt.Errorf("store: list summaries: %w", err)
+			return fmt.Errorf("store: list summaries: %w", err)
 		}
 		defer rows.Close()
 
-	var out []SummaryRow
 		for rows.Next() {
 			row, err := scanSummaryRow(rows)
 			if err != nil {
-			return nil, err
+				return err
 			}
 			out = append(out, row)
 		}
 		if err := rows.Err(); err != nil {
-		return nil, fmt.Errorf("store: iterate summaries: %w", err)
+			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,
+	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 nil, fmt.Errorf("store: get summary: %w", err)
+			return 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)
+				return fmt.Errorf("store: get summary: %w", err)
 			}
-		return nil, ErrNotFound
+			return nil
 		}
-	row, err := scanSummaryRow(rows)
+		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

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
+}

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,14 +145,9 @@ 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)
+	return s.withUser(ctx, sum.UserID, func(tx pgx.Tx) error {
-	if err != nil {
+		// Ensure the owning user exists (FK target). The store sink receives only
-		return fmt.Errorf("store: begin: %w", err)
+		// a Summary, so a minimal user row is enough at Stage 0.
-	}
-	defer tx.Rollback(ctx) //nolint:errcheck // no-op after Commit
-
-	// 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 {
@@ -147,20 +183,19 @@ func (s *Store) Deliver(ctx context.Context, sum domain.Summary) error {
 			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
+	})
 }
 
 // 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,
+	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 {
+			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,
+	seen := make(map[string]bool)
+	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 nil, fmt.Errorf("store: seen video ids: %w", err)
+			return 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)
+				return fmt.Errorf("store: scan video id: %w", err)
 			}
 			seen[id] = true
 		}
 		if err := rows.Err(); err != nil {
-		return nil, fmt.Errorf("store: iterate video ids: %w", err)
+			return fmt.Errorf("store: iterate video ids: %w", err)
+		}
+		return nil
+	}); err != nil {
+		return nil, err
 	}
 	return seen, nil
 }

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,25 +31,20 @@ 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 := 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.QueryRow(ctx,
 			`INSERT INTO videos (user_id, provider, provider_video_id, title, url, published_at)
 			 VALUES ($1, $2, $3, $4, $5, $6)
@@ -58,11 +55,11 @@ func (s *Store) UpsertVideo(ctx context.Context, v domain.Video) (string, error)
 			 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 fmt.Errorf("store: upsert video: %w", err)
 		}
-
+		return nil
-	if err := tx.Commit(ctx); err != nil {
+	}); err != nil {
-		return "", fmt.Errorf("store: commit: %w", err)
+		return "", err
 	}
 	return id, nil
 }