feat(store): video_connections table + RLS + connection store methods

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) <noreply@anthropic.com>
2026-06-03 16:12:34 +02:00
parent 7b4960e417
commit 0c9531a9b8
5 changed files with 254 additions and 2 deletions
@@ -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
+}
@@ -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)
+}
@@ -0,0 +1 @@
+DROP TABLE IF EXISTS video_connections;
@@ -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);
@@ -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
 }