From 0c9531a9b8618f8552532704dc053dbcf0f92929 Mon Sep 17 00:00:00 2001 From: Mathias Date: Wed, 3 Jun 2026 16:12:34 +0200 Subject: [PATCH] feat(store): video_connections table + RLS + connection store methods MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the video_connections table (data-model VIDEO_CONNECTION) with FORCE row-level security keyed off tapir.current_user_id, identical to migration 003's per-user isolation pattern — connections are user-owned data and must be isolated at the DB layer, not only by application WHERE clauses. Store methods (UpsertConnection / ConnectionsForUser / DeleteConnection) all route through withUser so RLS scopes every access. UpsertConnection is idempotent on (user_id, provider). The OAuth refresh token never lives here; token_ref is the opaque SecretStore reference. Extend the RLS isolation proof to cover video_connections: seeded per user, included in the deny-all + scoped-read assertions, and added to the cross-user write-invisibility and survivor checks. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/adapters/store/connections.go | 100 ++++++++++++++++ internal/adapters/store/connections_test.go | 111 ++++++++++++++++++ .../migrations/005_video_connections.down.sql | 1 + .../migrations/005_video_connections.up.sql | 30 +++++ internal/adapters/store/rls_test.go | 14 ++- 5 files changed, 254 insertions(+), 2 deletions(-) create mode 100644 internal/adapters/store/connections.go create mode 100644 internal/adapters/store/connections_test.go create mode 100644 internal/adapters/store/migrations/005_video_connections.down.sql create mode 100644 internal/adapters/store/migrations/005_video_connections.up.sql diff --git a/internal/adapters/store/connections.go b/internal/adapters/store/connections.go new file mode 100644 index 0000000..2f6b23f --- /dev/null +++ b/internal/adapters/store/connections.go @@ -0,0 +1,100 @@ +package store + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5" +) + +// Connection is one connected video account (data-model.md VIDEO_CONNECTION). +// TokenRef is the opaque SecretStore reference that resolves to the OAuth refresh +// token — never the token itself. ConnectedAt is set by the DB and is read-only +// on writes (UpsertConnection ignores it). +type Connection struct { + Provider string + ProviderAccount string + TokenRef string + Status string + ConnectedAt time.Time +} + +// UpsertConnection records (or refreshes) the user's connection to a provider, +// keyed on (user_id, provider): re-connecting the same provider overwrites the +// token_ref/status/account and bumps connected_at, never duplicating. Like every +// access in this package it routes through withUser, so RLS scopes the write to +// the calling user — a connection can only be written for the current user. +func (s *Store) UpsertConnection(ctx context.Context, userID string, c Connection) error { + return s.withUser(ctx, userID, func(tx pgx.Tx) error { + if _, err := tx.Exec(ctx, + `INSERT INTO video_connections + (user_id, provider, provider_account, token_ref, status) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (user_id, provider) DO UPDATE SET + provider_account = EXCLUDED.provider_account, + token_ref = EXCLUDED.token_ref, + status = EXCLUDED.status, + connected_at = now()`, + userID, c.Provider, nullIfEmpty(c.ProviderAccount), c.TokenRef, c.Status, + ); err != nil { + return fmt.Errorf("store: upsert connection: %w", err) + } + return nil + }) +} + +// ConnectionsForUser returns the user's connections, most-recently-connected +// first. Scoped by user_id via withUser: one user never sees another's. +func (s *Store) ConnectionsForUser(ctx context.Context, userID string) ([]Connection, error) { + var out []Connection + if err := s.withUser(ctx, userID, func(tx pgx.Tx) error { + rows, err := tx.Query(ctx, + `SELECT provider, COALESCE(provider_account, ''), token_ref, status, connected_at + FROM video_connections + WHERE user_id = $1 + ORDER BY connected_at DESC, provider`, + userID) + if err != nil { + return fmt.Errorf("store: connections for user: %w", err) + } + defer rows.Close() + + for rows.Next() { + var c Connection + if err := rows.Scan(&c.Provider, &c.ProviderAccount, &c.TokenRef, &c.Status, &c.ConnectedAt); err != nil { + return fmt.Errorf("store: scan connection: %w", err) + } + out = append(out, c) + } + if err := rows.Err(); err != nil { + return fmt.Errorf("store: iterate connections: %w", err) + } + return nil + }); err != nil { + return nil, err + } + return out, nil +} + +// DeleteConnection removes the user's connection to a provider. Deleting an +// absent connection is a no-op (no error). +func (s *Store) DeleteConnection(ctx context.Context, userID, provider string) error { + return s.withUser(ctx, userID, func(tx pgx.Tx) error { + if _, err := tx.Exec(ctx, + `DELETE FROM video_connections WHERE user_id = $1 AND provider = $2`, + userID, provider); err != nil { + return fmt.Errorf("store: delete connection: %w", err) + } + return nil + }) +} + +// nullIfEmpty maps "" to a SQL NULL so an unknown provider_account is stored as +// NULL (the column is nullable) rather than an empty string. +func nullIfEmpty(s string) *string { + if s == "" { + return nil + } + return &s +} diff --git a/internal/adapters/store/connections_test.go b/internal/adapters/store/connections_test.go new file mode 100644 index 0000000..ad62c80 --- /dev/null +++ b/internal/adapters/store/connections_test.go @@ -0,0 +1,111 @@ +package store_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + "gitea.d-ma.be/mathias/tapir/internal/adapters/store" +) + +// seedUserRow inserts a bare users row (FK target for a connection) as the +// superuser pool, which bypasses RLS. +func seedUserRow(t *testing.T, userID string) { + t.Helper() + _, err := rawPool(t).Exec(context.Background(), + `INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, userID) + require.NoError(t, err) +} + +func TestUpsertConnectionInsertsRow(t *testing.T) { + ctx := context.Background() + s := newStore(t) + p := rawPool(t) + resetDB(t, p) + seedUserRow(t, userA) + + require.NoError(t, s.UpsertConnection(ctx, userA, store.Connection{ + Provider: "youtube", + ProviderAccount: "chan@example.com", + TokenRef: "youtube/" + userA + "/refresh_token", + Status: "active", + })) + + conns, err := s.ConnectionsForUser(ctx, userA) + require.NoError(t, err) + require.Len(t, conns, 1) + require.Equal(t, "youtube", conns[0].Provider) + require.Equal(t, "chan@example.com", conns[0].ProviderAccount) + require.Equal(t, "youtube/"+userA+"/refresh_token", conns[0].TokenRef) + require.Equal(t, "active", conns[0].Status) + require.False(t, conns[0].ConnectedAt.IsZero(), "connected_at set by the DB default") +} + +func TestUpsertConnectionIsIdempotentOnUserProvider(t *testing.T) { + ctx := context.Background() + s := newStore(t) + p := rawPool(t) + resetDB(t, p) + seedUserRow(t, userA) + + require.NoError(t, s.UpsertConnection(ctx, userA, store.Connection{ + Provider: "youtube", TokenRef: "ref-1", Status: "active", + })) + // Re-connect the same provider: must update in place, not duplicate. + require.NoError(t, s.UpsertConnection(ctx, userA, store.Connection{ + Provider: "youtube", TokenRef: "ref-2", Status: "revoked", + })) + + var count int + require.NoError(t, p.QueryRow(ctx, + `SELECT count(*) FROM video_connections WHERE user_id = $1 AND provider = 'youtube'`, + userA).Scan(&count)) + require.Equal(t, 1, count, "second connect must update, not duplicate") + + conns, err := s.ConnectionsForUser(ctx, userA) + require.NoError(t, err) + require.Len(t, conns, 1) + require.Equal(t, "ref-2", conns[0].TokenRef, "token_ref overwritten") + require.Equal(t, "revoked", conns[0].Status, "status overwritten") +} + +func TestDeleteConnection(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + seedUserRow(t, userA) + + require.NoError(t, s.UpsertConnection(ctx, userA, store.Connection{ + Provider: "youtube", TokenRef: "ref", Status: "active", + })) + require.NoError(t, s.DeleteConnection(ctx, userA, "youtube")) + + conns, err := s.ConnectionsForUser(ctx, userA) + require.NoError(t, err) + require.Empty(t, conns) + + // Deleting an absent connection is a no-op, not an error. + require.NoError(t, s.DeleteConnection(ctx, userA, "youtube")) +} + +func TestConnectionsForUserIsScoped(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + seedUserRow(t, userA) + seedUserRow(t, userB) + + require.NoError(t, s.UpsertConnection(ctx, userA, store.Connection{ + Provider: "youtube", TokenRef: "a-ref", Status: "active", + })) + + // User B must not see user A's connection. + connsB, err := s.ConnectionsForUser(ctx, userB) + require.NoError(t, err) + require.Empty(t, connsB, "user B must not see user A's connections") + + connsA, err := s.ConnectionsForUser(ctx, userA) + require.NoError(t, err) + require.Len(t, connsA, 1) +} diff --git a/internal/adapters/store/migrations/005_video_connections.down.sql b/internal/adapters/store/migrations/005_video_connections.down.sql new file mode 100644 index 0000000..9f28d98 --- /dev/null +++ b/internal/adapters/store/migrations/005_video_connections.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS video_connections; diff --git a/internal/adapters/store/migrations/005_video_connections.up.sql b/internal/adapters/store/migrations/005_video_connections.up.sql new file mode 100644 index 0000000..ac9d710 --- /dev/null +++ b/internal/adapters/store/migrations/005_video_connections.up.sql @@ -0,0 +1,30 @@ +-- Migration 005: video_connections — a user's connected video account +-- (data-model.md VIDEO_CONNECTION). The OAuth refresh token never lives here; +-- token_ref is the opaque SecretStore reference that resolves to it. Revocation +-- flips status, it does not delete the row (history is kept). +-- +-- One connection per (user, provider): re-connecting the same provider upserts +-- in place (the connect flow's ON CONFLICT (user_id, provider) target). +CREATE TABLE video_connections ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + provider TEXT NOT NULL, + provider_account TEXT, + token_ref TEXT NOT NULL, + status TEXT NOT NULL, + connected_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT video_connections_user_provider_unique UNIQUE (user_id, provider) +); + +CREATE INDEX idx_video_connections_user_id ON video_connections(user_id); + +-- Per-user isolation, identical to migration 003's pattern: this is user-owned +-- data, so user A must never read or write user B's connections even with a wrong +-- application-level WHERE. ENABLE + FORCE so the table owner (tapir) is subject to +-- the policy too; the policy keys off the per-request GUC tapir.current_user_id +-- set by the store's withUser helper. An unset GUC yields NULL -> deny-all. +ALTER TABLE video_connections ENABLE ROW LEVEL SECURITY; +ALTER TABLE video_connections FORCE ROW LEVEL SECURITY; +CREATE POLICY video_connections_isolation ON video_connections + FOR ALL + USING (user_id = current_setting('tapir.current_user_id', true)::uuid); diff --git a/internal/adapters/store/rls_test.go b/internal/adapters/store/rls_test.go index 7eb8e07..b01176d 100644 --- a/internal/adapters/store/rls_test.go +++ b/internal/adapters/store/rls_test.go @@ -24,7 +24,7 @@ import ( // 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", + "users", "videos", "transcripts", "summaries", "summary_actions", "video_connections", } // allIsolatedTables adds sink_deliveries, whose ownership is derived from its @@ -74,6 +74,11 @@ func seedUser(t *testing.T, p *pgxpool.Pool, userID string) seeded { VALUES ($1, 'store', 'delivered')`, summaryID) require.NoError(t, err) + _, err = p.Exec(ctx, + `INSERT INTO video_connections (user_id, provider, token_ref, status) + VALUES ($1, 'youtube', $2, 'active')`, userID, "youtube/"+userID+"/refresh_token") + require.NoError(t, err) + return seeded{userID: userID, videoID: videoID, summaryID: summaryID} } @@ -180,9 +185,11 @@ func TestRLSEnforcesPerUserIsolation(t *testing.T) { {"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}, + {"update video_connections", `UPDATE video_connections SET token_ref = 'hacked' WHERE user_id = $1`, b.userID}, {"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}, + {"delete video_connections", `DELETE FROM video_connections WHERE user_id = $1`, b.userID}, } for _, w := range writes { require.Equal(t, int64(0), scopedRowsAffected(t, app, userA, w.sql, w.arg), @@ -197,16 +204,19 @@ func TestRLSEnforcesPerUserIsolation(t *testing.T) { `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 + var bSummaries, bActions, bDeliveries, bConnections 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.NoError(t, super.QueryRow(ctx, + `SELECT count(*) FROM video_connections WHERE user_id = $1 AND token_ref <> 'hacked'`, b.userID).Scan(&bConnections)) 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") + require.Equal(t, 1, bConnections, "A's writes must not have touched B's connection") _ = a // a's ids are seeded for the symmetric read assertions above }