mathias/tapir
1
0
Fork 0
Code Issues 25 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(store): route all DB access through withUser for structural scoping

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
mathias
2026-06-03 15:15:52 +02:00
co-authored by Claude Opus 4.8
parent 6775e5f53d
commit 7b139c2cd7
4 changed files with 231 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/adapters/store/actions.go
+51 -50
View File
@@ -3,6 +3,8 @@ package store
import ( import (
"context" "context"
"fmt" "fmt"
"github.com/jackc/pgx/v5"
) )
// allowedActions is the closed set of action verbs persisted in summary_actions. // 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 return err
} }
tx, err := s.pool.Begin(ctx) return s.withUser(ctx, userID, func(tx pgx.Tx) error {
if err != nil { if opposite, ok := oppositeAction[action]; ok {
return fmt.Errorf("store: begin set action: %w", err) if _, err := tx.Exec(ctx,
} `DELETE FROM summary_actions
defer tx.Rollback(ctx) //nolint:errcheck // no-op after Commit WHERE user_id = $1 AND video_id = $2 AND action = $3`,
userID, videoID, opposite); err != nil {
if opposite, ok := oppositeAction[action]; ok { return fmt.Errorf("store: clear opposite action: %w", err)
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, if _, err := tx.Exec(ctx,
`INSERT INTO summary_actions (user_id, video_id, action) `INSERT INTO summary_actions (user_id, video_id, action)
VALUES ($1, $2, $3) VALUES ($1, $2, $3)
ON CONFLICT (user_id, video_id, action) DO UPDATE SET acted_at = NOW()`, ON CONFLICT (user_id, video_id, action) DO UPDATE SET acted_at = NOW()`,
userID, videoID, action); err != nil { userID, videoID, action); err != nil {
return fmt.Errorf("store: set action: %w", err) return fmt.Errorf("store: set action: %w", err)
} }
return nil
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 // 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 { if err := validateAction(action); err != nil {
return err return err
} }
if _, err := s.pool.Exec(ctx, return s.withUser(ctx, userID, func(tx pgx.Tx) error {
`DELETE FROM summary_actions if _, err := tx.Exec(ctx,
WHERE user_id = $1 AND video_id = $2 AND action = $3`, `DELETE FROM summary_actions
userID, videoID, action); err != nil { WHERE user_id = $1 AND video_id = $2 AND action = $3`,
return fmt.Errorf("store: clear action: %w", err) userID, videoID, action); err != nil {
} return fmt.Errorf("store: clear action: %w", err)
return nil }
return nil
})
} }
// ActionsFor returns the active actions per video for the given user, keyed by // 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 { if len(videoIDs) == 0 {
return out, nil return out, nil
} }
rows, err := s.pool.Query(ctx, if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
`SELECT video_id, action FROM summary_actions rows, err := tx.Query(ctx,
WHERE user_id = $1 AND video_id = ANY($2) `SELECT video_id, action FROM summary_actions
ORDER BY video_id, action`, WHERE user_id = $1 AND video_id = ANY($2)
userID, videoIDs) ORDER BY video_id, action`,
if err != nil { userID, videoIDs)
return nil, fmt.Errorf("store: actions for: %w", err) if err != nil {
} 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)
} }
out[videoID] = append(out[videoID], action) defer rows.Close()
}
if err := rows.Err(); err != nil { for rows.Next() {
return nil, fmt.Errorf("store: iterate actions: %w", err) 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 return out, nil
} }
internal/adapters/store/reads.go
+52 -34
View File
@@ -66,27 +66,32 @@ func (s *Store) ListSummaries(ctx context.Context, userID string, limit int) ([]
if limit <= 0 { if limit <= 0 {
limit = 50 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 var out []SummaryRow
for rows.Next() { if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
row, err := scanSummaryRow(rows) rows, err := tx.Query(ctx,
selectSummary+`
WHERE s.user_id = $1
ORDER BY s.created_at DESC
LIMIT $2`,
userID, limit)
if err != nil { if err != nil {
return nil, err return fmt.Errorf("store: list summaries: %w", err)
} }
out = append(out, row) defer rows.Close()
}
if err := rows.Err(); err != nil { for rows.Next() {
return nil, fmt.Errorf("store: iterate summaries: %w", err) 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 { if err := s.attachActions(ctx, userID, out); err != nil {
return nil, err 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 // highlights and takeaways. Returns ErrNotFound when the user has no such
// summary. Scoped by user_id. // summary. Scoped by user_id.
func (s *Store) GetSummaryByVideo(ctx context.Context, userID, videoID string) (*SummaryRow, error) { func (s *Store) GetSummaryByVideo(ctx context.Context, userID, videoID string) (*SummaryRow, error) {
rows, err := s.pool.Query(ctx, var (
selectSummary+` row SummaryRow
WHERE s.user_id = $1 AND s.video_id = $2`, found bool
userID, videoID) )
if err != nil { if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
return nil, fmt.Errorf("store: get summary: %w", err) rows, err := tx.Query(ctx,
} selectSummary+`
defer rows.Close() WHERE s.user_id = $1 AND s.video_id = $2`,
userID, videoID)
if !rows.Next() { if err != nil {
if err := rows.Err(); err != nil { return fmt.Errorf("store: get summary: %w", err)
return nil, fmt.Errorf("store: get summary: %w", err)
} }
return nil, ErrNotFound defer rows.Close()
}
row, err := scanSummaryRow(rows) if !rows.Next() {
if err != nil { 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 return nil, err
} }
if !found {
return nil, ErrNotFound
}
holder := []SummaryRow{row} holder := []SummaryRow{row}
if err := s.attachActions(ctx, userID, holder); err != nil { if err := s.attachActions(ctx, userID, holder); err != nil {
return nil, err return nil, err
internal/adapters/store/store.go
+104 -64
View File
@@ -19,6 +19,7 @@ import (
"github.com/golang-migrate/migrate/v4" "github.com/golang-migrate/migrate/v4"
migratepgx "github.com/golang-migrate/migrate/v4/database/pgx/v5" migratepgx "github.com/golang-migrate/migrate/v4/database/pgx/v5"
"github.com/golang-migrate/migrate/v4/source/iofs" "github.com/golang-migrate/migrate/v4/source/iofs"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool" "github.com/jackc/pgx/v5/pgxpool"
_ "github.com/jackc/pgx/v5/stdlib" // register the "pgx" database/sql driver for migrate _ "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. // Name identifies this sink in delivery records.
func (s *Store) Name() string { return "store" } 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 // 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 // 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 // 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) 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.
} if _, err := tx.Exec(ctx,
defer tx.Rollback(ctx) //nolint:errcheck // no-op after Commit `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 var summaryID string
// Summary, so a minimal user row is enough at Stage 0. if err := tx.QueryRow(ctx,
if _, err := tx.Exec(ctx, `INSERT INTO summaries
`INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, (user_id, video_id, summary, highlights, takeaways, ai_provider, ai_model, fallback_used)
sum.UserID); err != nil { VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
return fmt.Errorf("store: upsert user: %w", err) 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.Exec(ctx,
if err := tx.QueryRow(ctx, `INSERT INTO sink_deliveries (summary_id, sink, status)
`INSERT INTO summaries VALUES ($1, 'store', 'delivered')
(user_id, video_id, summary, highlights, takeaways, ai_provider, ai_model, fallback_used) ON CONFLICT (summary_id, sink) DO UPDATE SET
VALUES ($1, $2, $3, $4, $5, $6, $7, $8) status = 'delivered',
ON CONFLICT (user_id, video_id) DO UPDATE SET detail = NULL,
summary = EXCLUDED.summary, updated_at = NOW()`,
highlights = EXCLUDED.highlights, summaryID); err != nil {
takeaways = EXCLUDED.takeaways, return fmt.Errorf("store: record delivery: %w", err)
ai_provider = EXCLUDED.ai_provider, }
ai_model = EXCLUDED.ai_model, return nil
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
} }
// HasSummary reports whether a summary already exists for (userID, videoID). // HasSummary reports whether a summary already exists for (userID, videoID).
// This is the per-video durable dedup check. // This is the per-video durable dedup check.
func (s *Store) HasSummary(ctx context.Context, userID, videoID string) (bool, error) { func (s *Store) HasSummary(ctx context.Context, userID, videoID string) (bool, error) {
var exists bool var exists bool
if err := s.pool.QueryRow(ctx, if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
`SELECT EXISTS(SELECT 1 FROM summaries WHERE user_id = $1 AND video_id = $2)`, return tx.QueryRow(ctx,
userID, videoID).Scan(&exists); err != nil { `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 false, fmt.Errorf("store: has summary: %w", err)
} }
return exists, nil 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. The watcher uses it to skip re-summarizing across restarts. Scoped by
// user_id, so one user never sees another's videos. // user_id, so one user never sees another's videos.
func (s *Store) SeenVideoIDs(ctx context.Context, userID string) (map[string]bool, error) { 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) seen := make(map[string]bool)
for rows.Next() { if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
var id string rows, err := tx.Query(ctx,
if err := rows.Scan(&id); err != nil { `SELECT video_id FROM summaries WHERE user_id = $1`, userID)
return nil, fmt.Errorf("store: scan video id: %w", err) if err != nil {
return fmt.Errorf("store: seen video ids: %w", err)
} }
seen[id] = true defer rows.Close()
}
if err := rows.Err(); err != nil { for rows.Next() {
return nil, fmt.Errorf("store: iterate video ids: %w", err) 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 return seen, nil
} }
internal/adapters/store/videos.go
+24 -27
View File
@@ -5,6 +5,8 @@ import (
"fmt" "fmt"
"time" "time"
"github.com/jackc/pgx/v5"
"gitea.d-ma.be/mathias/tapir/internal/domain" "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") 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) provider := string(v.Provider)
if provider == "" { if provider == "" {
provider = string(domain.ProviderYouTube) provider = string(domain.ProviderYouTube)
} }
var id string var id string
if err := tx.QueryRow(ctx, if err := s.withUser(ctx, v.UserID, func(tx pgx.Tx) error {
`INSERT INTO videos (user_id, provider, provider_video_id, title, url, published_at) // Ensure the owning user exists (FK target) — same as the Deliver path.
VALUES ($1, $2, $3, $4, $5, $6) if _, err := tx.Exec(ctx,
ON CONFLICT (user_id, provider, provider_video_id) DO UPDATE SET `INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`,
title = EXCLUDED.title, v.UserID); err != nil {
url = EXCLUDED.url, return fmt.Errorf("store: upsert user: %w", err)
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 := tx.Commit(ctx); err != nil { if err := tx.QueryRow(ctx,
return "", fmt.Errorf("store: commit: %w", err) `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 return id, nil
} }