test: add postgres admin state contract fixture

.github/workflows/ci.yml

@@ -32,3 +32,19 @@ jobs:
           go-version-file: go.mod
           cache: true
       - run: go test -race -timeout 60s ./internal/...
+
+  integration:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          go-version-file: go.mod
+          cache: true
+      - name: Redis activity contract
+        shell: pwsh
+        run: ./scripts/test-redis.ps1
+      - name: PostgreSQL admin-state contract
+        shell: pwsh
+        run: ./scripts/test-postgres.ps1

deploy/compose_test.go

@@ -57,6 +57,25 @@ func TestRedisFixtureScriptUsesDedicatedComposeProject(t *testing.T) {
 	}
 }
 
+func TestPostgresFixtureScriptUsesDedicatedComposeProject(t *testing.T) {
+	payload, err := os.ReadFile("../scripts/test-postgres.ps1")
+	if err != nil {
+		t.Fatalf("read test-postgres.ps1: %v", err)
+	}
+	script := string(payload)
+	for _, required := range []string{
+		`-p $composeProject`,
+		`up -d --wait --wait-timeout 60 postgres`,
+		`PROXY_POOL_TEST_POSTGRES_URL`,
+		`go test -count=1 -tags=integration -timeout 60s ./internal/adapters/postgresadmin/...`,
+		`down --volumes --remove-orphans`,
+	} {
+		if !strings.Contains(script, required) {
+			t.Errorf("test-postgres.ps1 missing %q", required)
+		}
+	}
+}
+
 func TestLocalRedisIsExplicitlyEphemeral(t *testing.T) {
 	document := loadComposeDocument(t)
 	redis, ok := document.Services["redis"]

internal/adapters/postgresadmin/contract_integration_test.go

@@ -0,0 +1,83 @@
+//go:build integration
+
+package postgresadmin
+
+import (
+	"context"
+	"slices"
+	"strings"
+	"testing"
+
+	"proxy-pool/internal/domain/adminstate"
+	"proxy-pool/internal/domain/adminstate/contracttest"
+)
+
+func TestPostgresAdminStateContract(t *testing.T) {
+	contracttest.Run(t, func(t *testing.T) adminstate.Store {
+		return newPostgresTestFixture(t).Store
+	})
+}
+
+func TestMigrationsAreIdempotentAndBounded(t *testing.T) {
+	fixture := newPostgresTestFixture(t)
+	if err := ApplyMigrations(t.Context(), fixture.Pool); err != nil {
+		t.Fatalf("ApplyMigrations(second run): %v", err)
+	}
+
+	rows, err := fixture.Pool.Query(context.Background(), `
+		SELECT table_name
+		FROM information_schema.tables
+		WHERE table_schema = $1 AND table_type = 'BASE TABLE'
+		ORDER BY table_name`, fixture.Schema)
+	if err != nil {
+		t.Fatalf("query migrated tables: %v", err)
+	}
+	defer rows.Close()
+	tables := make([]string, 0, 6)
+	for rows.Next() {
+		var name string
+		if err := rows.Scan(&name); err != nil {
+			t.Fatalf("scan migrated table: %v", err)
+		}
+		tables = append(tables, name)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatalf("iterate migrated tables: %v", err)
+	}
+	want := []string{
+		"admin_audit_log",
+		"admin_outbox",
+		"config_revisions",
+		"control_revisions",
+		"routing_admin_state",
+		"upstream_admin_state",
+	}
+	if !slices.Equal(tables, want) {
+		t.Fatalf("migrated tables = %v, want %v", tables, want)
+	}
+
+	columnRows, err := fixture.Pool.Query(context.Background(), `
+		SELECT table_name, column_name
+		FROM information_schema.columns
+		WHERE table_schema = $1`, fixture.Schema)
+	if err != nil {
+		t.Fatalf("query migrated columns: %v", err)
+	}
+	defer columnRows.Close()
+	for columnRows.Next() {
+		var table string
+		var column string
+		if err := columnRows.Scan(&table, &column); err != nil {
+			t.Fatalf("scan migrated column: %v", err)
+		}
+		qualified := strings.ToLower(table + "." + column)
+		for _, forbidden := range []string{"proxy", "credential", "extraction", "worker_owner", "idempotency"} {
+			if strings.Contains(qualified, forbidden) {
+				t.Fatalf("PostgreSQL management boundary contains forbidden column %s", qualified)
+			}
+		}
+	}
+	if err := columnRows.Err(); err != nil {
+		t.Fatalf("iterate migrated columns: %v", err)
+	}
+}

internal/adapters/postgresadmin/migrate.go

@@ -0,0 +1,47 @@
+package postgresadmin
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"proxy-pool/internal/domain/adminstate"
+)
+
+// ApplyMigrations runs every embedded idempotent migration on one physical connection.
+func ApplyMigrations(ctx context.Context, pool *pgxpool.Pool) error {
+	if err := contextError(ctx); err != nil {
+		return err
+	}
+	if isNil(pool) {
+		return adminstate.ErrInvalidCommand
+	}
+	connection, err := pool.Acquire(ctx)
+	if err != nil {
+		return migrationError(ctx, err)
+	}
+	defer connection.Release()
+
+	for _, migration := range Migrations() {
+		if _, err := connection.Exec(ctx, migration.SQL); err != nil {
+			rollbackContext, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+			_, _ = connection.Exec(rollbackContext, "ROLLBACK")
+			return migrationError(ctx, err)
+		}
+	}
+	return nil
+}
+
+func migrationError(ctx context.Context, err error) error {
+	if ctx != nil && ctx.Err() != nil {
+		return ctx.Err()
+	}
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return err
+	}
+	return fmt.Errorf("postgresadmin: apply migrations: %w", adminstate.ErrUnavailable)
+}

internal/adapters/postgresadmin/migrations_test.go

@@ -1,12 +1,29 @@
 package postgresadmin
 
 import (
+	"context"
+	"errors"
 	"regexp"
 	"sort"
 	"strings"
 	"testing"
+
+	"proxy-pool/internal/domain/adminstate"
 )
 
+func TestApplyMigrationsValidatesContextAndPool(t *testing.T) {
+	t.Parallel()
+
+	if err := ApplyMigrations(context.Background(), nil); !errors.Is(err, adminstate.ErrInvalidCommand) {
+		t.Fatalf("ApplyMigrations(nil) error = %v, want ErrInvalidCommand", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if err := ApplyMigrations(ctx, nil); !errors.Is(err, context.Canceled) {
+		t.Fatalf("ApplyMigrations(canceled) error = %v, want context.Canceled", err)
+	}
+}
+
 func TestMigrationsAreOrderedTransactionalAndImmutable(t *testing.T) {
 	t.Parallel()
 	migrations := Migrations()

internal/adapters/postgresadmin/rollback_integration_test.go

@@ -0,0 +1,142 @@
+//go:build integration
+
+package postgresadmin
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"proxy-pool/internal/domain/adminstate"
+)
+
+func TestMutationRollsBackWhenAuditInsertFails(t *testing.T) {
+	fixture := newPostgresTestFixture(t)
+	now := integrationNow()
+	commitIntegrationConfig(t, fixture.Store, now)
+	installRejectInsertTrigger(t, fixture, "admin_audit_log", "reject_admin_audit")
+
+	_, err := fixture.Store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{
+		RequestID: "req-audit-failure", Actor: integrationActor(), OccurredAt: now.Add(time.Second),
+		Name: "provider-a", Enabled: false,
+	})
+	if !errors.Is(err, adminstate.ErrUnavailable) {
+		t.Fatalf("SetUpstreamEnabled(audit failure) error = %v, want ErrUnavailable", err)
+	}
+	assertFailedMutationLeftBaseline(t, fixture)
+
+	dropRejectInsertTrigger(t, fixture, "admin_audit_log", "reject_admin_audit")
+	result, err := fixture.Store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{
+		RequestID: "req-after-audit-failure", Actor: integrationActor(), OccurredAt: now.Add(2 * time.Second),
+		Name: "provider-a", Enabled: false,
+	})
+	if err != nil || !result.Changed || result.Revision != 2 {
+		t.Fatalf("SetUpstreamEnabled(after rollback) = %+v, %v", result, err)
+	}
+}
+
+func TestMutationRollsBackWhenOutboxInsertFails(t *testing.T) {
+	fixture := newPostgresTestFixture(t)
+	now := integrationNow()
+	commitIntegrationConfig(t, fixture.Store, now)
+	installRejectInsertTrigger(t, fixture, "admin_outbox", "reject_admin_outbox")
+
+	_, err := fixture.Store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{
+		RequestID: "req-outbox-failure", Actor: integrationActor(), OccurredAt: now.Add(time.Second),
+		Name: "provider-a", Enabled: false,
+	})
+	if !errors.Is(err, adminstate.ErrUnavailable) {
+		t.Fatalf("SetUpstreamEnabled(outbox failure) error = %v, want ErrUnavailable", err)
+	}
+	assertFailedMutationLeftBaseline(t, fixture)
+
+	dropRejectInsertTrigger(t, fixture, "admin_outbox", "reject_admin_outbox")
+	result, err := fixture.Store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{
+		RequestID: "req-after-outbox-failure", Actor: integrationActor(), OccurredAt: now.Add(2 * time.Second),
+		Name: "provider-a", Enabled: false,
+	})
+	if err != nil || !result.Changed || result.Revision != 2 {
+		t.Fatalf("SetUpstreamEnabled(after rollback) = %+v, %v", result, err)
+	}
+}
+
+func assertFailedMutationLeftBaseline(t *testing.T, fixture postgresTestFixture) {
+	t.Helper()
+	snapshot, err := fixture.Store.Snapshot(context.Background())
+	if err != nil || snapshot.Revision != 1 || !integrationUpstreamEnabled(snapshot, "provider-a") {
+		t.Fatalf("Snapshot(after failed mutation) = %+v, %v", snapshot, err)
+	}
+	audits, err := fixture.Store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 10})
+	if err != nil || len(audits) != 1 {
+		t.Fatalf("ReadAudit(after failed mutation) = %+v, %v", audits, err)
+	}
+	for table, want := range map[string]int{"control_revisions": 1, "admin_outbox": 1} {
+		var count int
+		if err := fixture.Pool.QueryRow(context.Background(), "SELECT COUNT(*) FROM "+table).Scan(&count); err != nil {
+			t.Fatalf("count %s: %v", table, err)
+		}
+		if count != want {
+			t.Fatalf("%s row count = %d, want %d", table, count, want)
+		}
+	}
+}
+
+func installRejectInsertTrigger(t *testing.T, fixture postgresTestFixture, table, trigger string) {
+	t.Helper()
+	function := trigger + "_fn"
+	statement := "CREATE FUNCTION " + function + `() RETURNS trigger LANGUAGE plpgsql AS $$
+		BEGIN RAISE EXCEPTION 'injected management write failure'; END
+	$$;
+	CREATE TRIGGER ` + trigger + " BEFORE INSERT ON " + table +
+		" FOR EACH ROW EXECUTE FUNCTION " + function + "()"
+	if _, err := fixture.Pool.Exec(context.Background(), statement); err != nil {
+		t.Fatalf("install %s trigger: %v", trigger, err)
+	}
+}
+
+func dropRejectInsertTrigger(t *testing.T, fixture postgresTestFixture, table, trigger string) {
+	t.Helper()
+	statement := "DROP TRIGGER " + trigger + " ON " + table + "; DROP FUNCTION " + trigger + "_fn()"
+	if _, err := fixture.Pool.Exec(context.Background(), statement); err != nil {
+		t.Fatalf("drop %s trigger: %v", trigger, err)
+	}
+}
+
+func commitIntegrationConfig(t *testing.T, store adminstate.Store, now time.Time) {
+	t.Helper()
+	result, err := store.CommitConfig(context.Background(), adminstate.CommitConfigCommand{
+		RequestID: "req-config", Actor: integrationActor(), OccurredAt: now,
+		ConfigVersion: "cfg-1", Checksum: strings.Repeat("a", adminstate.SHA256HexBytes),
+		Source: "configs/proxy-pool.yaml",
+		Upstreams: []adminstate.UpstreamDefinition{
+			{Name: "provider-a", Enabled: true},
+			{Name: "provider-b", Enabled: true},
+		},
+		Routings: []adminstate.RoutingDefinition{{
+			Name: "checkout", Enabled: true, Upstreams: []string{"provider-a", "provider-b"},
+			CurrentUpstream: "provider-a",
+		}},
+	})
+	if err != nil || !result.Changed || result.Revision != 1 {
+		t.Fatalf("CommitConfig() = %+v, %v", result, err)
+	}
+}
+
+func integrationActor() adminstate.Actor {
+	return adminstate.Actor{ID: "admin-a", SourceIP: "192.0.2.10"}
+}
+
+func integrationNow() time.Time {
+	return time.Date(2026, 7, 30, 10, 0, 0, 0, time.UTC)
+}
+
+func integrationUpstreamEnabled(snapshot adminstate.Snapshot, name string) bool {
+	for _, upstream := range snapshot.Upstreams {
+		if upstream.Name == name {
+			return upstream.Enabled
+		}
+	}
+	return false
+}

internal/adapters/postgresadmin/testpostgres_test.go

@@ -0,0 +1,98 @@
+//go:build integration
+
+package postgresadmin
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"proxy-pool/internal/domain/adminstate"
+)
+
+var postgresTestSchemaSequence atomic.Uint64
+
+type postgresTestFixture struct {
+	Store   adminstate.Store
+	Pool    *pgxpool.Pool
+	Schema  string
+	Cleanup func()
+}
+
+func newPostgresTestFixture(t *testing.T) postgresTestFixture {
+	t.Helper()
+	postgresURL := os.Getenv("PROXY_POOL_TEST_POSTGRES_URL")
+	if postgresURL == "" {
+		t.Skip("PROXY_POOL_TEST_POSTGRES_URL is not set")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	adminConfig, err := pgxpool.ParseConfig(postgresURL)
+	if err != nil {
+		t.Fatalf("parse PROXY_POOL_TEST_POSTGRES_URL: %v", err)
+	}
+	adminPool, err := pgxpool.NewWithConfig(ctx, adminConfig)
+	if err != nil {
+		t.Fatalf("connect test PostgreSQL: %v", err)
+	}
+	if err := adminPool.Ping(ctx); err != nil {
+		adminPool.Close()
+		t.Fatalf("ping test PostgreSQL: %v", err)
+	}
+
+	schema := fmt.Sprintf("it_%d_%d_%d", os.Getpid(), time.Now().UnixNano(), postgresTestSchemaSequence.Add(1))
+	quotedSchema := pgx.Identifier{schema}.Sanitize()
+	if _, err := adminPool.Exec(ctx, "CREATE SCHEMA "+quotedSchema); err != nil {
+		adminPool.Close()
+		t.Fatalf("create isolated PostgreSQL schema: %v", err)
+	}
+
+	testConfig, err := pgxpool.ParseConfig(postgresURL)
+	if err != nil {
+		adminPool.Close()
+		t.Fatalf("parse isolated PostgreSQL config: %v", err)
+	}
+	testConfig.ConnConfig.RuntimeParams["search_path"] = schema
+	testPool, err := pgxpool.NewWithConfig(ctx, testConfig)
+	if err != nil {
+		_, _ = adminPool.Exec(context.Background(), "DROP SCHEMA "+quotedSchema+" CASCADE")
+		adminPool.Close()
+		t.Fatalf("connect isolated PostgreSQL schema: %v", err)
+	}
+	if err := ApplyMigrations(ctx, testPool); err != nil {
+		testPool.Close()
+		_, _ = adminPool.Exec(context.Background(), "DROP SCHEMA "+quotedSchema+" CASCADE")
+		adminPool.Close()
+		t.Fatalf("apply PostgreSQL migrations: %v", err)
+	}
+	store, err := New(testPool)
+	if err != nil {
+		testPool.Close()
+		_, _ = adminPool.Exec(context.Background(), "DROP SCHEMA "+quotedSchema+" CASCADE")
+		adminPool.Close()
+		t.Fatalf("construct PostgreSQL adapter: %v", err)
+	}
+
+	var cleanupOnce sync.Once
+	cleanup := func() {
+		cleanupOnce.Do(func() {
+			testPool.Close()
+			cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer cleanupCancel()
+			if _, err := adminPool.Exec(cleanupCtx, "DROP SCHEMA "+quotedSchema+" CASCADE"); err != nil {
+				t.Errorf("drop isolated PostgreSQL schema: %v", err)
+			}
+			adminPool.Close()
+		})
+	}
+	t.Cleanup(cleanup)
+	return postgresTestFixture{Store: store, Pool: testPool, Schema: schema, Cleanup: cleanup}
+}

scripts/test-postgres.ps1

@@ -0,0 +1,34 @@
+$ErrorActionPreference = "Stop"
+
+$repositoryRoot = Split-Path -Parent $PSScriptRoot
+$composeFile = Join-Path $repositoryRoot "deploy/docker-compose.test.yml"
+$composeProject = "proxy-pool-postgres-test"
+$previousPostgresURL = [Environment]::GetEnvironmentVariable("PROXY_POOL_TEST_POSTGRES_URL", "Process")
+
+try {
+    docker compose -p $composeProject -f $composeFile up -d --wait --wait-timeout 60 postgres
+    if ($LASTEXITCODE -ne 0) {
+        throw "starting PostgreSQL test fixture failed with exit code $LASTEXITCODE"
+    }
+
+    $env:PROXY_POOL_TEST_POSTGRES_URL = "postgres://proxy_pool_test:proxy-pool-test@127.0.0.1:15432/proxy_pool_test?sslmode=disable"
+    Push-Location $repositoryRoot
+    try {
+        go test -count=1 -tags=integration -timeout 60s ./internal/adapters/postgresadmin/...
+        if ($LASTEXITCODE -ne 0) {
+            throw "PostgreSQL integration tests failed with exit code $LASTEXITCODE"
+        }
+    }
+    finally {
+        Pop-Location
+    }
+}
+finally {
+    if ($null -eq $previousPostgresURL) {
+        Remove-Item Env:PROXY_POOL_TEST_POSTGRES_URL -ErrorAction SilentlyContinue
+    }
+    else {
+        $env:PROXY_POOL_TEST_POSTGRES_URL = $previousPostgresURL
+    }
+    docker compose -p $composeProject -f $composeFile down --volumes --remove-orphans
+}