test: Move event_outbox to tenant schema in party cascade test (#2140)

* test: Move event_outbox to tenant schema in party cascade test

Create event_outbox in the tenant schema to match the production schema
placement and the pattern in internal-account/service/outbox_test.go.
Adds tenant_id column and uses qualified schema name.

Set database-level search_path via ALTER DATABASE so all connection pool
connections (including the outbox worker's) resolve unqualified table
names to the tenant schema. A session-level SET search_path only affects
a single connection, which caused the load test to receive 0 Kafka events
under concurrent load.

* test: Address review feedback on event_outbox tenant schema setup

- Use pq.QuoteIdentifier instead of %q for SQL identifier quoting,
  consistent with the other CREATE TABLE statements in the same function
- Remove &events.EventOutbox{} from SetupPostgres: the public-schema
  auto-migrate is superseded by the explicit tenant-schema CREATE TABLE
  below, eliminating the duplicate table

---------

Co-authored-by: Ben Coombs <bjcoombs@users.noreply.github.com>

Author: bjcoombs

services/party/service/kafka_cascade_integration_test.go

@@ -101,11 +101,11 @@ func setupKafkaCascadeTest(t *testing.T) *kafkaCascadeTestEnv {
 	// Create the party.controlled.v1 topic
 	createKafkaTopic(t, broker, topics.PartyControlledV1)
 
-	// Setup CockroachDB with party + event_outbox tables
+	// Setup CockroachDB with party and audit tables in the public schema.
+	// event_outbox is created explicitly in the tenant schema below.
 	db, dbCleanup := testdb.SetupPostgres(t, []interface{}{
 		&persistence.PartyEntity{},
 		&audit.AuditOutbox{},
-		&events.EventOutbox{},
 	})
 
 	tid := tenant.TenantID(testTenantID)
@@ -157,17 +157,28 @@ func setupKafkaCascadeTest(t *testing.T) *kafkaCascadeTestEnv {
 		event_payload BYTEA NOT NULL,
 		correlation_id VARCHAR(100),
 		causation_id VARCHAR(100),
-		status VARCHAR(20) NOT NULL DEFAULT 'pending',
+		status VARCHAR(20) NOT NULL,
 		topic VARCHAR(200) NOT NULL,
 		partition_key VARCHAR(200),
-		created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
-		processed_at TIMESTAMP WITH TIME ZONE,
+		created_at TIMESTAMPTZ NOT NULL,
+		processed_at TIMESTAMPTZ,
 		retry_count INTEGER NOT NULL DEFAULT 0,
 		last_error TEXT,
-		service_name VARCHAR(100) NOT NULL
+		service_name VARCHAR(100) NOT NULL,
+		tenant_id VARCHAR(100) NOT NULL
 	)`, pq.QuoteIdentifier(schemaName))).Error)
 
-	require.NoError(t, db.Exec(fmt.Sprintf("SET search_path TO %s", pq.QuoteIdentifier(schemaName))).Error)
+	// Set search_path at the database level so all pool connections (including the outbox
+	// worker's) resolve unqualified table names to the tenant schema. A session-level
+	// SET search_path only affects a single pool connection; the load test needs all
+	// connections to share the same schema routing.
+	require.NoError(t, db.Exec(fmt.Sprintf("ALTER DATABASE test_db SET search_path TO %s, public", pq.QuoteIdentifier(schemaName))).Error)
+	// Close idle connections so they reconnect and pick up the new database default.
+	sqlDB, err := db.DB()
+	require.NoError(t, err)
+	sqlDB.SetMaxIdleConns(0) // closes idle connections immediately
+	// Also set it on the current session for any in-flight use of this connection.
+	require.NoError(t, db.Exec(fmt.Sprintf("SET search_path TO %s, public", pq.QuoteIdentifier(schemaName))).Error)
 
 	ctx := tenant.WithTenant(bgCtx, tid)