test: validate atomic RLS changes through Supabase

docs/declarative-row-security.md

@@ -223,10 +223,11 @@ this table-scoped work.
 3. **Execute transitions atomically.** The Go executor now locks, derives, applies,
    and verifies one table's RLS state in one bounded transaction. Mixed changes and
    policy relation dependencies remain unsupported. CLI integration is a follow-up.
-4. **Prove application behavior.** Extend the local Supabase harness with real
-   authenticated and anonymous requests, two users, allowed and denied writes,
-   and interrupted transitions. Then validate hosted connection and privilege
-   boundaries on a disposable project before claiming hosted support.
+4. **Prove application behavior.** The local Supabase harness now checks real
+   PostgREST requests from two authenticated users and anonymous callers, allowed
+   and denied writes, changed visibility, and rollback after a cancelled apply.
+   Hosted connection and privilege validation remains a follow-up on a disposable
+   project; local results do not establish hosted support.
 
 The [inspection tests](../pkg/schemadiff/row_security_integration_test.go),
 [round-trip tests](../pkg/schemadiff/row_security_roundtrip_integration_test.go), and
@@ -241,3 +242,10 @@ Hosted validation is not a prerequisite for the local steps, nor replaced by the
 PostgreSQL's [`pg_policy` catalog](https://www.postgresql.org/docs/current/catalog-pg-policy.html)
 and [`CREATE POLICY` reference](https://www.postgresql.org/docs/current/sql-createpolicy.html)
 define the policy fields and behavior.
+
+The [RLS API tests](../integration/supabase/row_security_api_test.go) and
+[cancellation test](../integration/supabase/row_security_api_rollback_test.go)
+exercise the atomic executor on the pinned Supabase stack through application
+requests. They run automatically in the existing Supabase compatibility CI job.
+Tokens are fixture-signed; signup/login flows and Realtime authorization changes
+are outside these tests.

docs/supabase.md

@@ -186,7 +186,9 @@ hosted-project or complete Supabase stack test.
 | Desired plan containing a column removal | Refused with `destructive-change`; no safe prefix applied | [Whole-plan admission](../integration/supabase/failure_test.go) |
 | Lock contention, null rows during `SET NOT NULL`, and duplicate rows during a unique index build | Typed outcomes and durable database state verified; tenant API access preserved | [Failure paths](../integration/supabase/failure_test.go) |
 | API and Realtime after each copy-and-swap refusal | Tenant reads and new INSERT events still worked on the original sockets | [Service continuity](../integration/supabase/continuity_test.go) |
-| Enable RLS through the schema-change entry point | Refused with `unsupported-statement` | [Refusals](../integration/supabase/schema_test.go) |
+| Apply a complete RLS declaration through the Go API | PostgREST enforces tenant reads and writes, anonymous denial, changed visibility, and default deny; repeated apply is a no-op | [RLS application behavior](../integration/supabase/row_security_api_test.go) |
+| Cancel an RLS apply after a live policy is dropped | Transaction rollback restores the catalog and previous API access, including allowed and denied writes | [RLS rollback](../integration/supabase/row_security_api_rollback_test.go) |
+| Enable RLS through the statement/CLI entry point | Refused with `unsupported-statement` | [Refusals](../integration/supabase/schema_test.go) |
 | Supavisor session endpoint | Column addition and concurrent index succeeded; session timeouts verified | [Execution](../integration/supabase/services_test.go), [timeouts](../pkg/dbconn/supabase_integration_test.go) |
 | Supavisor transaction endpoint | Refused with `ErrNoSessionAffinity`, even with named prepared statements disabled | [Pooler boundary](../pkg/dbconn/supabase_integration_test.go) |
 | PostgREST after direct/session schema changes | New columns became available through automatic schema-cache reload | [API and tenants](../integration/supabase/services_test.go) |
@@ -210,9 +212,10 @@ outcome; they do not assume every unsuccessful change rolls back completely.
 
 ## What to keep in mind
 
-- RLS declarations can be exported and compared, but policy execution remains
-  unsupported. Table-only files do not compare access rules. Grants and roles
-  remain separately managed; see [declarative RLS](declarative-row-security.md)
+- RLS declarations can be exported, compared, and applied through the
+  [atomic Go API](atomic-row-security.md). CLI execution remains unsupported.
+  Table-only files do not compare access rules. Grants and roles remain separately
+  managed; see [declarative RLS](declarative-row-security.md)
 - Qualify extension types and functions outside `public`, such as
   `extensions.citext`. When pg-sprite inspects a desired schema file, it
   uses a separate workspace with its own search path. In policy expressions,

integration/supabase/README.md

@@ -43,6 +43,23 @@ and verify those leftovers; they do not assume every failure rolls back all work
 while a column addition and concurrent index build run with ongoing writes.
 Preserving publication membership alone would not prove event delivery.
 
+[row_security_api_test.go](row_security_api_test.go) applies complete policy definitions
+through the Go executor, then makes real PostgREST requests as two authenticated
+users and as an anonymous caller. Table grants deliberately allow these operations:
+the policy rules decide which rows are visible and which writes succeed.
+The fixture waits for Auth to initialize `auth.uid()` for PostgREST JWT claims,
+then waits for PostgREST to discover the table. PostgreSQL readiness alone does
+not establish either condition. Empty
+results for an unauthorized update or delete mean no rows were changed; a denied
+insert or ownership reassignment returns PostgreSQL error `42501` through the API.
+The tests also cover changed visibility, removing the last policy, and repeat apply.
+
+[row_security_api_rollback_test.go](row_security_api_rollback_test.go) cancels the
+executor immediately after PostgreSQL confirms a live `DROP POLICY`. It verifies
+catalog rollback and the same allowed and denied API access afterward. The tracer
+only triggers cancellation; all DDL and rollback run against the real database.
+These tests do not prove Realtime behavior during RLS changes or hosted support.
+
 ## Scope of the evidence
 
 The suite uses real local Supabase services and fixture-signed user tokens.

integration/supabase/row_security_api_helpers_test.go

@@ -0,0 +1,131 @@
+package supabase_test
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/block/pg-sprite/pkg/executor"
+	"github.com/block/pg-sprite/pkg/statement"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func rlsAPITable(t *testing.T, name string) *pgxpool.Pool {
+	t.Helper()
+	pool := fixture(t)
+	waitForRLSAuth(t, pool)
+	table := newTable(t, pool, name)
+	// Grant table access to both roles so refusals prove RLS, not missing grants.
+	execSQL(t, pool, "GRANT SELECT, INSERT, UPDATE, DELETE ON "+table+" TO authenticated, anon")
+	execSQL(t, pool, "INSERT INTO "+table+` VALUES
+     (1, '00000000-0000-0000-0000-000000000001', 'first'),
+     (2, '00000000-0000-0000-0000-000000000002', 'second')`)
+	execSQL(t, pool, "NOTIFY pgrst, 'reload schema'")
+	bearer := token(t, 1)
+	const cacheDeadline = 30 * time.Second
+	const cachePoll = 200 * time.Millisecond
+	require.Eventually(t, func() bool {
+		_, err := get(t.Context(), "http://127.0.0.1:55439/"+name+"?select=id", bearer)
+		return err == nil
+	}, cacheDeadline, cachePoll, "PostgREST must discover the fixture table")
+	return pool
+}
+
+func applyAPIRLS(t *testing.T, pool *pgxpool.Pool, sql string) executor.RowSecurityReport {
+	t.Helper()
+	desired, err := statement.ParseDesiredWithRowSecurity(sql)
+	require.NoError(t, err)
+	report, err := executor.ExecuteRowSecurity(t.Context(), pool, "public", desired, executor.Budget{LockTimeout: 100 * time.Millisecond, StatementTimeout: 5 * time.Second})
+	require.NoError(t, err)
+	return report
+}
+
+func rlsRequest(t *testing.T, method, path string, tenant int, payload string) (int, []byte) {
+	t.Helper()
+	req, err := http.NewRequestWithContext(t.Context(), method, "http://127.0.0.1:55439/"+path, bytes.NewBufferString(payload))
+	require.NoError(t, err)
+	if tenant != 0 {
+		req.Header.Set("Authorization", "Bearer "+token(t, tenant))
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Prefer", "return=representation")
+	client := http.Client{Timeout: 2 * time.Second}
+	response, err := client.Do(req)
+	require.NoError(t, err)
+	body, readErr := io.ReadAll(io.LimitReader(response.Body, 1<<20))
+	closeErr := response.Body.Close()
+	require.NoError(t, readErr)
+	require.NoError(t, closeErr)
+	return response.StatusCode, body
+}
+
+func assertRLSRows(t *testing.T, status int, body []byte, wantStatus int, ids ...int) {
+	t.Helper()
+	require.Equal(t, wantStatus, status, "%s", body)
+	var rows []struct {
+		ID int `json:"id"`
+	}
+	require.NoError(t, json.Unmarshal(body, &rows))
+	actual := make([]int, 0, len(rows))
+	for _, row := range rows {
+		actual = append(actual, row.ID)
+	}
+	assert.Equal(t, append([]int{}, ids...), actual)
+}
+
+func assertRLSRead(t *testing.T, name string, tenant int, ids ...int) {
+	t.Helper()
+	status, body := rlsRequest(t, http.MethodGet, name+"?select=id&order=id", tenant, "")
+	assertRLSRows(t, status, body, http.StatusOK, ids...)
+}
+
+func assertRLSInsertDenied(t *testing.T, name string, tenant, id, owner int) {
+	t.Helper()
+	status, body := rlsRequest(t, http.MethodPost, name, tenant, fmt.Sprintf(`{"id":%d,"owner_id":%q,"body":"denied"}`, id, tenantID(owner)))
+	wantStatus := http.StatusForbidden
+	if tenant == 0 {
+		wantStatus = http.StatusUnauthorized
+	}
+	assertRLSDenied(t, status, body, wantStatus)
+}
+
+func assertRLSDenied(t *testing.T, status int, body []byte, wantStatus int) {
+	t.Helper()
+	require.Equal(t, wantStatus, status, "%s", body)
+	var result struct {
+		Code string `json:"code"`
+	}
+	require.NoError(t, json.Unmarshal(body, &result))
+	assert.Equal(t, "42501", result.Code)
+}
+
+// Auth initializes the JWT-aware helper after PostgreSQL itself is healthy.
+// Probe its behavior without changing the function or leaking session settings.
+func waitForRLSAuth(t *testing.T, pool *pgxpool.Pool) {
+	t.Helper()
+	const authDeadline = 30 * time.Second
+	const authPoll = 200 * time.Millisecond
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		tx, err := pool.Begin(t.Context())
+		if !assert.NoError(c, err) {
+			return
+		}
+		defer func() { _ = tx.Rollback(context.WithoutCancel(t.Context())) }()
+		_, err = tx.Exec(t.Context(), "SELECT set_config('request.jwt.claims', $1, true)", fmt.Sprintf(`{"sub":%q}`, tenantID(1)))
+		if !assert.NoError(c, err) {
+			return
+		}
+		var subject string
+		if !assert.NoError(c, tx.QueryRow(t.Context(), "SELECT COALESCE(auth.uid()::text, '')").Scan(&subject)) {
+			return
+		}
+		assert.Equal(c, tenantID(1), subject, "Auth must initialize auth.uid() for PostgREST JWT claims")
+	}, authDeadline, authPoll)
+}

integration/supabase/row_security_api_rollback_test.go

@@ -0,0 +1,84 @@
+package supabase_test
+
+import (
+	"context"
+	"net/http"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/block/pg-sprite/pkg/executor"
+	"github.com/block/pg-sprite/pkg/schemadiff"
+	"github.com/block/pg-sprite/pkg/statement"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Cancel only after the server has successfully dropped a live policy. This
+// exercises partial work inside a real transaction, not a pre-execution refusal.
+type cancelAfterRLSDrop struct {
+	cancel  context.CancelFunc
+	dropSQL string
+	reached atomic.Bool
+}
+
+type rlsDropQueryKey struct{}
+
+func (c *cancelAfterRLSDrop) TraceQueryStart(ctx context.Context, _ *pgx.Conn, data pgx.TraceQueryStartData) context.Context {
+	return context.WithValue(ctx, rlsDropQueryKey{}, data.SQL == c.dropSQL)
+}
+func (c *cancelAfterRLSDrop) TraceQueryEnd(ctx context.Context, _ *pgx.Conn, data pgx.TraceQueryEndData) {
+	// Match the fully qualified live statement, not a DROP in a scratch schema.
+	liveDrop, _ := ctx.Value(rlsDropQueryKey{}).(bool)
+	if liveDrop && data.Err == nil && data.CommandTag.String() == "DROP POLICY" {
+		c.reached.Store(true)
+		c.cancel()
+	}
+}
+
+func TestAtomicRLSAPICancellationPreservesAccess(t *testing.T) {
+	const name = "pgsprite_rls_api_rollback"
+	pool := rlsAPITable(t, name)
+	before, err := schemadiff.Introspect(t.Context(), pool, "public", name)
+	require.NoError(t, err)
+	assertRLSRead(t, name, 1, 1)
+	assertRLSRead(t, name, 2, 2)
+	assertRLSRead(t, name, 0)
+	assertRLSInsertDenied(t, name, 1, 90, 2)
+	desired, err := statement.ParseDesiredWithRowSecurity(`CREATE TABLE pgsprite_rls_api_rollback (
+     id int PRIMARY KEY,
+     owner_id uuid NOT NULL,
+     body text NOT NULL
+ );
+ ALTER TABLE pgsprite_rls_api_rollback ENABLE ROW LEVEL SECURITY;
+ CREATE POLICY nobody ON pgsprite_rls_api_rollback
+     FOR SELECT TO authenticated USING (false);`)
+	require.NoError(t, err)
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	trace := &cancelAfterRLSDrop{
+		cancel:  cancel,
+		dropSQL: `DROP POLICY "own_rows" ON "public"."pgsprite_rls_api_rollback"`,
+	}
+	cfg := pool.Config()
+	cfg.ConnConfig.Tracer = trace
+	changing, err := pgxpool.NewWithConfig(t.Context(), cfg)
+	require.NoError(t, err)
+	defer changing.Close()
+	_, err = executor.ExecuteRowSecurity(ctx, changing, "public", desired, executor.Budget{LockTimeout: 100 * time.Millisecond, StatementTimeout: 5 * time.Second})
+	require.True(t, trace.reached.Load(), "cancellation must happen after live DROP POLICY succeeds")
+	require.ErrorIs(t, err, context.Canceled)
+	after, err := schemadiff.Introspect(t.Context(), pool, "public", name)
+	require.NoError(t, err)
+	assert.Equal(t, before, after)
+	assertRLSRead(t, name, 1, 1)
+	assertRLSRead(t, name, 2, 2)
+	assertRLSRead(t, name, 0)
+	assertRLSInsertDenied(t, name, 1, 91, 2)
+	assertRLSInsertDenied(t, name, 0, 92, 1)
+	// The original write policy survives too, not just its read behavior.
+	status, body := rlsRequest(t, http.MethodPost, name+"?select=id", 1, `{"id":3,"owner_id":"00000000-0000-0000-0000-000000000001","body":"after rollback"}`)
+	assertRLSRows(t, status, body, http.StatusCreated, 3)
+}