fix: repair final 6 integration failures + advisory-lock bug

Five test-side fixes exposed by the now-complete schema / first CI run of the suite:
- EmailVerification: verification tokens are hashed at rest (pkg/tokenhash), so
  assert the hash, not the raw token (same as SessionLifecycle).
- OutboxStore: jsonb round-trips through Postgres's canonical form (space after
  colon); compare with JSONEq instead of byte equality.
- RLSAudit_SentinelPolicyShape: pg_policies renders parsed expressions with
  explicit ::text casts; match that normalized shape in the sentinel fragment.
- TxManager_Reentrant: the inner write used a raw pool.Exec, which grabs its own
  connection and commits independently — route it through DBFromContext so it
  joins the (re-entrant) transaction and rolls back with the outer.
- RLS_TenantIsolation: newTenantUser built emails from a truncated UUIDv7 prefix;
  UUIDv7's leading bits are the ms timestamp, so users minted in the same tick
  collided on the unique email. Use the full UUID.

One production fix:
- PartitionManager advisory lock ran pg_try_advisory_lock via pool.QueryRow, so
  the session lock landed on an arbitrary pooled connection that was handed
  straight back, and pg_advisory_unlock ran on a possibly-different connection
  (silent no-op → leaked lock, and re-entrant TRUE for a second caller sharing
  the pool). Hold a dedicated pgxpool.Conn for the lock's lifetime, guarded by a
  mutex, and unlock+release on that same connection.

Author: Fr1ngg

internal/adapter/postgres/identity_repo_test.go

@@ -217,7 +217,8 @@ func TestIdentityRepo_EmailVerification(t *testing.T) {
 	assert.Equal(t, verification.ID, got.ID)
 	assert.Equal(t, verification.UserID, got.UserID)
 	assert.Equal(t, verification.Email, got.Email)
-	assert.Equal(t, verification.Token, got.Token)
+	// Verification tokens are hashed at rest (pkg/tokenhash), like refresh tokens.
+	assert.Equal(t, tokenhash.Hash(verification.Token), got.Token)
 
 	// Not found
 	_, err = repo.GetEmailVerification(ctx, "nonexistent-token")

internal/adapter/postgres/outbox_repo_test.go

@@ -72,7 +72,9 @@ func TestOutboxStore(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, events, 1)
 	assert.Equal(t, "subscription.activated", events[0].EventType)
-	assert.Equal(t, payload, events[0].Payload)
+	// jsonb round-trips through Postgres's canonical form (a space after the
+	// colon), so compare semantically rather than byte-for-byte.
+	assert.JSONEq(t, string(payload), string(events[0].Payload))
 	assert.NotEmpty(t, events[0].ID)
 	assert.False(t, events[0].CreatedAt.IsZero())
 }

internal/adapter/postgres/partition_manager.go

@@ -84,6 +84,15 @@ type PartitionManager struct {
 	logger    *slog.Logger
 	lookahead int           // quarters ahead to ensure
 	retention time.Duration // 0 = no cleanup
+
+	// lockMu guards lockConn. A session-level advisory lock lives on the
+	// connection that acquired it, so the lock must be held on a dedicated
+	// connection for its whole lifetime — pool.QueryRow would run the lock on
+	// an arbitrary pooled connection and hand it straight back, leaving the
+	// lock stranded on a random connection and pg_advisory_unlock running on a
+	// different one (a silent no-op → leaked lock).
+	lockMu   sync.Mutex
+	lockConn *pgxpool.Conn
 }
 
 // NewPartitionManager creates a PartitionManager that pre-creates future
@@ -130,26 +139,57 @@ func (pm *PartitionManager) Run(ctx context.Context) {
 	}
 }
 
-// tryAdvisoryLock attempts to acquire a session-level advisory lock.
-// Returns true if the lock was acquired, false if another pod holds it
-// or the query fails.
+// tryAdvisoryLock attempts to acquire a session-level advisory lock on a
+// dedicated connection held for the lock's lifetime. Returns true if the lock
+// was acquired, false if another session holds it or the query fails.
 func (pm *PartitionManager) tryAdvisoryLock(ctx context.Context) bool {
-	var acquired bool
-	err := pm.pool.QueryRow(ctx, "SELECT pg_try_advisory_lock($1)", partitionManagerLockID).Scan(&acquired)
+	pm.lockMu.Lock()
+	defer pm.lockMu.Unlock()
+
+	if pm.lockConn != nil {
+		// This manager already holds the lock; do not stack a re-entrant
+		// acquisition (session advisory locks nest, and a single unlock would
+		// then leave the lock held).
+		return false
+	}
+
+	conn, err := pm.pool.Acquire(ctx)
 	if err != nil {
+		pm.logger.Warn("partition manager: advisory lock connection acquire failed", slog.Any("error", err))
+		return false
+	}
+
+	var acquired bool
+	if err := conn.QueryRow(ctx, "SELECT pg_try_advisory_lock($1)", partitionManagerLockID).Scan(&acquired); err != nil {
 		pm.logger.Warn("partition manager: advisory lock query failed", slog.Any("error", err))
+		conn.Release()
+		return false
+	}
+	if !acquired {
+		conn.Release()
 		return false
 	}
-	return acquired
+
+	pm.lockConn = conn
+	return true
 }
 
 // releaseAdvisoryLock releases the session-level advisory lock previously
-// acquired by tryAdvisoryLock. Errors are logged but not propagated.
+// acquired by tryAdvisoryLock, unlocking on the SAME connection that holds it
+// and returning that connection to the pool. Errors are logged but not
+// propagated. A no-op if this manager does not currently hold the lock.
 func (pm *PartitionManager) releaseAdvisoryLock(ctx context.Context) {
-	_, err := pm.pool.Exec(ctx, "SELECT pg_advisory_unlock($1)", partitionManagerLockID)
-	if err != nil {
+	pm.lockMu.Lock()
+	defer pm.lockMu.Unlock()
+
+	if pm.lockConn == nil {
+		return
+	}
+	if _, err := pm.lockConn.Exec(ctx, "SELECT pg_advisory_unlock($1)", partitionManagerLockID); err != nil {
 		pm.logger.Warn("partition manager: advisory unlock failed", slog.Any("error", err))
 	}
+	pm.lockConn.Release()
+	pm.lockConn = nil
 }
 
 // ensure delegates to OutboxRepository.EnsurePartitions to pre-create future

internal/adapter/postgres/rls_audit_test.go

@@ -146,8 +146,11 @@ func TestRLSAudit_SentinelPolicyShape(t *testing.T) {
 	pool := setupAuditDB(t)
 	ctx := context.Background()
 
-	// e.g. current_setting('app.tenant_id', true) = '*'  — sentinel from the named const.
-	sentinelFragment := "current_setting('app.tenant_id', true) = '" + sentinelGUC + "'"
+	// pg_policies renders the parsed expression in Postgres's canonical form,
+	// which adds explicit ::text casts to the literal and the GUC argument. Match
+	// that normalized shape, e.g.:
+	//   (current_setting('app.tenant_id'::text, true) = '*'::text)
+	sentinelFragment := "current_setting('app.tenant_id'::text, true) = '" + sentinelGUC + "'::text"
 
 	for _, tbl := range phaseCTenantTables {
 		policyName := tbl.policy

internal/adapter/postgres/rls_test.go

@@ -67,7 +67,10 @@ func newTenantUser(t *testing.T, tenantID string) *identity.PlatformUser {
 
 	return &identity.PlatformUser{
 		ID:            uuid.Must(uuid.NewV7()).String(),
-		Email:         fmt.Sprintf("user-%s@test.com", uuid.Must(uuid.NewV7()).String()[:8]),
+		// Full UUID (not a truncated prefix): UUIDv7's leading bits are the
+		// millisecond timestamp, so several users minted in the same tick would
+		// share a truncated prefix and collide on the unique email constraint.
+		Email:         fmt.Sprintf("user-%s@test.com", uuid.Must(uuid.NewV7()).String()),
 		PasswordHash:  "$2a$10$abcdefghijklmnopqrstuuABCDEFGHIJKLMNOPQRSTUVWXYZ012",
 		EmailVerified: false,
 		Role:          identity.RoleCustomer,

internal/adapter/postgres/txmanager_reentrant_test.go

@@ -46,7 +46,10 @@ func TestTxManager_Reentrant(t *testing.T) {
 	err := tm.RunInTx(ctx, func(outerCtx context.Context) error {
 		// Inner RunInTx: must join the outer tx (re-entrant path).
 		innerErr := tm.RunInTx(outerCtx, func(innerCtx context.Context) error {
-			_, execErr := pool.Exec(innerCtx,
+			// Write through the context-bound handle so the INSERT participates in
+			// the (re-entrant) transaction. A raw pool.Exec would grab its own
+			// connection and commit independently, defeating the test's premise.
+			_, execErr := postgres.DBFromContext(innerCtx, pool).Exec(innerCtx,
 				`INSERT INTO identity.platform_users (id, email, password_hash, role, email_verified)
 				 VALUES ($1, $2, 'hash', 'customer', false)`,
 				userID, "reentrant@test.com",