easyai-ai-gateway/apps/api/internal/store/identity_pairing_integration_test.go

package store

import (
	"context"
	"errors"
	"os"
	"path/filepath"
	"runtime"
	"strings"
	"testing"
	"time"

	"github.com/easyai/easyai-ai-gateway/apps/api/internal/identity"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

func TestIdentityPairingCancellationKeepsOlderInactivePairingBlockedUntilEveryCleanupCompletes(t *testing.T) {
	db := newIdentityPairingPostgresTestStore(t)
	ctx := context.Background()

	type pairingFixture struct {
		revisionID  string
		pairingID   string
		machineRef  string
		sessionRef  string
		createdAt   time.Time
		displayName string
	}
	now := time.Now().UTC()
	fixtures := []pairingFixture{
		{
			revisionID: uuid.NewString(), pairingID: uuid.NewString(),
			machineRef: "identity-machine-older", sessionRef: "identity-session-older",
			createdAt: now.Add(-time.Minute), displayName: "older",
		},
		{
			revisionID: uuid.NewString(), pairingID: uuid.NewString(),
			machineRef: "identity-machine-latest", sessionRef: "identity-session-latest",
			createdAt: now, displayName: "latest",
		},
	}

	for _, fixture := range fixtures {
		if _, err := db.pool.Exec(ctx, `
INSERT INTO gateway_identity_configuration_revisions (
  id,state,auth_center_url,local_tenant_key,public_base_url,web_base_url,
  machine_credential_ref,session_encryption_key_ref,created_at,updated_at
) VALUES ($1::uuid,'draft','https://auth.test.example','default','https://gateway.test.example',
  'https://gateway-web.test.example',$2,$3,$4,$4)`,
			fixture.revisionID, fixture.machineRef, fixture.sessionRef, fixture.createdAt); err != nil {
			t.Fatalf("seed %s inactive revision: %v", fixture.displayName, err)
		}
		if _, err := db.CreateIdentityPairingExchange(ctx, identity.PairingExchange{
			ID: fixture.pairingID, RevisionID: fixture.revisionID, RemoteExchangeID: uuid.NewString(),
			ExchangeTokenRef: "identity-exchange-" + fixture.pairingID,
			Status:           identity.PairingCredentialsSaved, CleanupStatus: identity.PairingCleanupNone,
			RemoteVersion: 4, ExpiresAt: now.Add(time.Hour), LastTraceID: "trace-start-" + fixture.displayName,
		}); err != nil {
			t.Fatalf("seed %s pairing: %v", fixture.displayName, err)
		}
	}
	orphanRevisionID := uuid.NewString()
	if _, err := db.pool.Exec(ctx, `
INSERT INTO gateway_identity_configuration_revisions (
  id,state,auth_center_url,local_tenant_key,public_base_url,web_base_url,created_at,updated_at
) VALUES ($1::uuid,'draft','https://auth.test.example','default','https://orphan-api.test.example',
  'https://orphan-web.test.example',$2,$2)`, orphanRevisionID, now.Add(time.Minute)); err != nil {
		t.Fatalf("seed orphan revision: %v", err)
	}
	canonical, err := db.LatestInactiveIdentityConfigurationRevision(ctx)
	if err != nil || canonical.ID != fixtures[1].revisionID {
		t.Fatalf("canonical inactive revision=%#v error=%v, want latest exchange-backed revision", canonical, err)
	}

	blocked, err := db.IdentityPairingStartBlocked(ctx)
	if err != nil || !blocked {
		t.Fatalf("two inactive pairings blocked=%t error=%v, want blocked", blocked, err)
	}

	latest := fixtures[1]
	latestCancelled, err := db.CancelIdentityPairingExchange(ctx, latest.pairingID, 1, "trace-cancel-latest", "audit-cancel-latest")
	if err != nil {
		t.Fatalf("cancel latest pairing: %v", err)
	}
	latestReplay, err := db.CancelIdentityPairingExchange(ctx, latest.pairingID, 1, "trace-replayed", "audit-replayed")
	if err != nil {
		t.Fatalf("replay latest cancellation with original version: %v", err)
	}
	if latestReplay.Status != identity.PairingCancelled || latestReplay.CleanupStatus != identity.PairingCleanupPending ||
		latestReplay.Version != latestCancelled.Version || latestCancelled.CancelledAt == nil || latestReplay.CancelledAt == nil ||
		!latestReplay.CancelledAt.Equal(*latestCancelled.CancelledAt) {
		t.Fatalf("cancellation replay changed persisted intent: first=%#v replay=%#v", latestCancelled, latestReplay)
	}
	latestCleaned, err := db.CompleteIdentityPairingCleanup(ctx, latest.pairingID, latestCancelled.Version)
	if err != nil {
		t.Fatalf("complete latest cleanup: %v", err)
	}
	if latestCleaned.CleanupStatus != identity.PairingCleanupCompleted || latestCleaned.CleanupCompletedAt == nil {
		t.Fatalf("latest cleanup state=%#v", latestCleaned)
	}
	assertIdentityRevisionSecretRefs(t, ctx, db, latest.revisionID, "", "")
	assertIdentityRevisionSecretRefs(t, ctx, db, fixtures[0].revisionID, fixtures[0].machineRef, fixtures[0].sessionRef)

	blocked, err = db.IdentityPairingStartBlocked(ctx)
	if err != nil || !blocked {
		t.Fatalf("older pairing was ignored after latest cleanup: blocked=%t error=%v", blocked, err)
	}
	canonical, err = db.LatestInactiveIdentityConfigurationRevision(ctx)
	if err != nil || canonical.ID != fixtures[0].revisionID {
		t.Fatalf("canonical inactive revision after latest cleanup=%#v error=%v, want older blocker", canonical, err)
	}

	older := fixtures[0]
	olderCancelled, err := db.CancelIdentityPairingExchange(ctx, older.pairingID, 1, "trace-cancel-older", "audit-cancel-older")
	if err != nil {
		t.Fatalf("cancel older pairing: %v", err)
	}
	olderReplay, err := db.CancelIdentityPairingExchange(ctx, older.pairingID, 1, "trace-replayed", "audit-replayed")
	if err != nil || olderReplay.Version != olderCancelled.Version {
		t.Fatalf("replay older cancellation: replay=%#v error=%v", olderReplay, err)
	}
	if _, err := db.CompleteIdentityPairingCleanup(ctx, older.pairingID, olderCancelled.Version); err != nil {
		t.Fatalf("complete older cleanup: %v", err)
	}
	assertIdentityRevisionSecretRefs(t, ctx, db, older.revisionID, "", "")

	blocked, err = db.IdentityPairingStartBlocked(ctx)
	if err != nil || blocked {
		t.Fatalf("all inactive pairing cleanup blocked=%t error=%v, want unblocked", blocked, err)
	}
	canonical, err = db.LatestInactiveIdentityConfigurationRevision(ctx)
	if err != nil || canonical.ID != fixtures[1].revisionID || canonical.ID == orphanRevisionID {
		t.Fatalf("canonical cleaned revision=%#v error=%v, orphan draft must stay hidden", canonical, err)
	}
}

func TestIdentityPairingStartReservationSerializesAndRejectsStaleCommit(t *testing.T) {
	db := newIdentityPairingPostgresTestStore(t)
	ctx := context.Background()
	firstAttempt := uuid.NewString()
	secondAttempt := uuid.NewString()

	if err := db.ReserveIdentityPairingStart(ctx, firstAttempt, time.Now().Add(time.Minute)); err != nil {
		t.Fatalf("reserve first pairing start: %v", err)
	}
	if err := db.ReserveIdentityPairingStart(ctx, secondAttempt, time.Now().Add(time.Minute)); !errors.Is(err, identity.ErrPairingInProgress) {
		t.Fatalf("second pairing reservation error=%v, want in-progress conflict", err)
	}
	if _, err := db.pool.Exec(ctx, `UPDATE gateway_identity_pairing_start_reservation SET expires_at=now()-interval '1 second'
WHERE attempt_id=$1::uuid`, firstAttempt); err != nil {
		t.Fatalf("expire first pairing reservation: %v", err)
	}
	if err := db.ReserveIdentityPairingStart(ctx, secondAttempt, time.Now().Add(time.Minute)); err != nil {
		t.Fatalf("take over expired pairing reservation: %v", err)
	}
	staleRevision := identity.Revision{
		ID: uuid.NewString(), State: identity.RevisionDraft, SchemaVersion: 1, AuthCenterURL: "https://auth.test.example",
		RolePrefix: "gateway.", LocalTenantKey: "default", PublicBaseURL: "https://gateway.test.example",
		WebBaseURL: "https://gateway-web.test.example", Scopes: []string{}, Capabilities: []string{},
		SessionIdleSeconds: 1800, SessionAbsoluteSeconds: 28800, SessionRefreshSeconds: 60,
	}
	if _, err := db.CommitIdentityPairingStart(ctx, staleRevision, identity.PairingExchange{
		ID: firstAttempt, RevisionID: staleRevision.ID, RemoteExchangeID: uuid.NewString(),
		ExchangeTokenRef: "identity-exchange-" + firstAttempt, Status: identity.PairingMetadataPending,
		CleanupStatus: identity.PairingCleanupNone, RemoteVersion: 1, ExpiresAt: time.Now().Add(time.Minute),
	}); !errors.Is(err, identity.ErrPairingInProgress) {
		t.Fatalf("stale start commit error=%v, want in-progress conflict", err)
	}
	validRevision := staleRevision
	validRevision.ID = uuid.NewString()
	tokenReference := "identity-exchange-" + secondAttempt
	if err := db.QueueIdentitySecretCleanup(ctx, tokenReference, time.Now().Add(10*time.Minute)); err != nil {
		t.Fatalf("stage replacement exchange token reference: %v", err)
	}
	committed, err := db.CommitIdentityPairingStart(ctx, validRevision, identity.PairingExchange{
		ID: secondAttempt, RevisionID: validRevision.ID, RemoteExchangeID: uuid.NewString(),
		ExchangeTokenRef: tokenReference, Status: identity.PairingMetadataPending,
		CleanupStatus: identity.PairingCleanupNone, RemoteVersion: 1, ExpiresAt: time.Now().Add(time.Minute),
	})
	if err != nil {
		t.Fatalf("commit replacement pairing: %v", err)
	}
	if err := db.ReleaseIdentityPairingStart(ctx, secondAttempt); err != nil {
		t.Fatalf("release paired reservation: %v", err)
	}
	var reservationState string
	if err := db.pool.QueryRow(ctx, `SELECT state FROM gateway_identity_pairing_start_reservation
WHERE attempt_id=$1::uuid`, secondAttempt).Scan(&reservationState); err != nil || reservationState != "paired" {
		t.Fatalf("paired reservation state=%q error=%v", reservationState, err)
	}
	blocked, err := db.IdentityPairingStartBlocked(ctx)
	if err != nil || !blocked {
		t.Fatalf("committed pairing did not retain singleton reservation: blocked=%t error=%v", blocked, err)
	}
	cancelled, err := db.CancelIdentityPairingExchange(ctx, committed.ID, committed.Version, "trace-cancel", "audit-cancel")
	if err != nil {
		t.Fatalf("cancel replacement pairing: %v", err)
	}
	if _, err := db.CompleteIdentityPairingCleanup(ctx, committed.ID, cancelled.Version); err != nil {
		t.Fatalf("complete replacement pairing cleanup: %v", err)
	}
	blocked, err = db.IdentityPairingStartBlocked(ctx)
	if err != nil || blocked {
		t.Fatalf("completed cleanup retained singleton reservation: blocked=%t error=%v", blocked, err)
	}
}

func TestIdentityPairingStartRejectsAnyActiveConfigurationBeforeReservation(t *testing.T) {
	db := newIdentityPairingPostgresTestStore(t)
	ctx := context.Background()
	if _, err := db.pool.Exec(ctx, `INSERT INTO gateway_identity_configuration_revisions (
  id,state,auth_center_url,local_tenant_key,public_base_url,web_base_url
) VALUES ($1::uuid,'active','https://auth.test.example','default','https://gateway.test.example',
  'https://gateway-web.test.example')`, uuid.NewString()); err != nil {
		t.Fatalf("seed active identity revision: %v", err)
	}

	err := db.ReserveIdentityPairingStart(ctx, uuid.NewString(), time.Now().Add(time.Minute))
	if !errors.Is(err, identity.ErrActiveConfigurationHandoffRequired) {
		t.Fatalf("active identity configuration reservation error=%v", err)
	}
	var reservations int
	if err := db.pool.QueryRow(ctx, `SELECT count(*) FROM gateway_identity_pairing_start_reservation`).Scan(&reservations); err != nil {
		t.Fatal(err)
	}
	if reservations != 0 {
		t.Fatalf("active credential guard left %d start reservations", reservations)
	}
}

func TestConcurrentIdentityPairingStartReservationsHaveSingleWinner(t *testing.T) {
	db := newIdentityPairingPostgresTestStore(t)
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	attempts := []string{uuid.NewString(), uuid.NewString()}
	ready := make(chan struct{}, len(attempts))
	start := make(chan struct{})
	type reservationResult struct {
		attemptID string
		err       error
	}
	results := make(chan reservationResult, len(attempts))
	for _, attemptID := range attempts {
		go func() {
			ready <- struct{}{}
			<-start
			results <- reservationResult{
				attemptID: attemptID,
				err:       db.ReserveIdentityPairingStart(ctx, attemptID, time.Now().Add(5*time.Minute)),
			}
		}()
	}
	for range attempts {
		<-ready
	}
	close(start)

	var winner string
	conflicts := 0
	for range attempts {
		result := <-results
		switch {
		case result.err == nil:
			if winner != "" {
				t.Fatalf("multiple concurrent reservations succeeded: %s and %s", winner, result.attemptID)
			}
			winner = result.attemptID
		case errors.Is(result.err, identity.ErrPairingInProgress):
			conflicts++
		default:
			t.Fatalf("reserve concurrent pairing start %s: %v", result.attemptID, result.err)
		}
	}
	if winner == "" || conflicts != 1 {
		t.Fatalf("concurrent reservation winner=%q conflicts=%d, want one winner and one conflict", winner, conflicts)
	}

	var persistedAttemptID, state string
	var reservations int
	if err := db.pool.QueryRow(ctx, `SELECT count(*),COALESCE(max(attempt_id::text),''),COALESCE(max(state),'')
FROM gateway_identity_pairing_start_reservation`).Scan(&reservations, &persistedAttemptID, &state); err != nil {
		t.Fatalf("read concurrent reservation result: %v", err)
	}
	if reservations != 1 || persistedAttemptID != winner || state != "starting" {
		t.Fatalf("persisted reservations=%d attempt=%q state=%q, want winner %q in starting state",
			reservations, persistedAttemptID, state, winner)
	}
}

func TestConcurrentIdentityActivationAndPairingReservationRemainMutuallyExclusive(t *testing.T) {
	db := newIdentityPairingPostgresTestStore(t)
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	seedIdentityActivationPrerequisites(t, ctx, db, "default")
	revisionID := seedIdentityLifecycleRevision(t, ctx, db, identity.RevisionValidated, "default")
	attemptID := uuid.NewString()

	gate, err := db.pool.Begin(ctx)
	if err != nil {
		t.Fatalf("begin identity lifecycle gate: %v", err)
	}
	defer gate.Rollback(context.Background())
	if err := lockIdentityConfigurationLifecycle(ctx, gate); err != nil {
		t.Fatalf("lock identity lifecycle gate: %v", err)
	}

	reserveResult := make(chan error, 1)
	go func() {
		reserveResult <- db.ReserveIdentityPairingStart(ctx, attemptID, time.Now().Add(5*time.Minute))
	}()
	waitForIdentityLifecycleLockWaiters(t, ctx, db, 1)

	type activationResult struct {
		revision identity.Revision
		err      error
	}
	activateResult := make(chan activationResult, 1)
	go func() {
		revision, _, activateErr := db.ActivateIdentityRevision(ctx, revisionID, 1, "trace-activate", "audit-activate")
		activateResult <- activationResult{revision: revision, err: activateErr}
	}()
	waitForIdentityLifecycleLockWaiters(t, ctx, db, 2)
	if err := gate.Commit(ctx); err != nil {
		t.Fatalf("release identity lifecycle gate: %v", err)
	}

	reserveErr := <-reserveResult
	activated := <-activateResult
	if reserveErr != nil {
		t.Fatalf("first queued pairing reservation failed: %v", reserveErr)
	}

	var activeRevisions, startingReservations int
	if err := db.pool.QueryRow(ctx, `SELECT
  (SELECT count(*) FROM gateway_identity_configuration_revisions WHERE state='active'),
  (SELECT count(*) FROM gateway_identity_pairing_start_reservation WHERE state='starting')`).Scan(
		&activeRevisions, &startingReservations,
	); err != nil {
		t.Fatalf("read activation and reservation invariant: %v", err)
	}
	if !errors.Is(activated.err, identity.ErrPairingInProgress) || activeRevisions != 0 || startingReservations != 1 {
		t.Fatalf("racing activation error=%v revision_state=%q left active=%d starting reservations=%d; "+
			"want pairing-in-progress, active=0, reservation=1",
			activated.err, activated.revision.State, activeRevisions, startingReservations)
	}
}

func TestConcurrentIdentityPairingCommitRejectsActiveRevisionCommittedAheadOfLifecycleLock(t *testing.T) {
	db := newIdentityPairingPostgresTestStore(t)
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	attemptID := uuid.NewString()
	if err := db.ReserveIdentityPairingStart(ctx, attemptID, time.Now().Add(5*time.Minute)); err != nil {
		t.Fatalf("reserve pairing start before commit race: %v", err)
	}
	tokenReference := "identity-exchange-" + attemptID
	if err := db.QueueIdentitySecretCleanup(ctx, tokenReference, time.Now().Add(10*time.Minute)); err != nil {
		t.Fatalf("stage exchange token reference before commit race: %v", err)
	}
	draft := identity.Revision{
		ID: uuid.NewString(), State: identity.RevisionDraft, SchemaVersion: 1, AuthCenterURL: "https://auth.test.example",
		RolePrefix: "gateway.", LocalTenantKey: "default", PublicBaseURL: "https://gateway.test.example",
		WebBaseURL: "https://gateway-web.test.example", Scopes: []string{}, Capabilities: []string{},
		SessionIdleSeconds: 1800, SessionAbsoluteSeconds: 28800, SessionRefreshSeconds: 60,
	}
	exchange := identity.PairingExchange{
		ID: attemptID, RevisionID: draft.ID, RemoteExchangeID: uuid.NewString(), ExchangeTokenRef: tokenReference,
		Status: identity.PairingMetadataPending, CleanupStatus: identity.PairingCleanupNone,
		RemoteVersion: 1, ExpiresAt: time.Now().Add(5 * time.Minute),
	}

	gate, err := db.pool.Begin(ctx)
	if err != nil {
		t.Fatalf("begin identity lifecycle gate: %v", err)
	}
	defer gate.Rollback(context.Background())
	if err := lockIdentityConfigurationLifecycle(ctx, gate); err != nil {
		t.Fatalf("lock identity lifecycle gate: %v", err)
	}

	commitResult := make(chan error, 1)
	go func() {
		_, commitErr := db.CommitIdentityPairingStart(ctx, draft, exchange)
		commitResult <- commitErr
	}()
	waitForIdentityLifecycleLockWaiters(t, ctx, db, 1)
	activeID := uuid.NewString()
	if _, err := gate.Exec(ctx, `INSERT INTO gateway_identity_configuration_revisions (
  id,state,auth_center_url,local_tenant_key,public_base_url,web_base_url
) VALUES ($1::uuid,'active','https://auth.test.example','default','https://active-gateway.test.example',
  'https://active-gateway-web.test.example')`, activeID); err != nil {
		t.Fatalf("commit active revision ahead of pairing commit: %v", err)
	}
	if err := gate.Commit(ctx); err != nil {
		t.Fatalf("release identity lifecycle gate with active revision: %v", err)
	}

	commitErr := <-commitResult
	if !errors.Is(commitErr, identity.ErrActiveConfigurationHandoffRequired) {
		t.Fatalf("pairing commit after active revision error=%v, want active handoff required", commitErr)
	}
	var activeRevisions, draftRevisions, exchanges, startingReservations int
	if err := db.pool.QueryRow(ctx, `SELECT
  (SELECT count(*) FROM gateway_identity_configuration_revisions WHERE state='active'),
  (SELECT count(*) FROM gateway_identity_configuration_revisions WHERE id=$1::uuid),
  (SELECT count(*) FROM gateway_identity_onboarding_exchanges WHERE id=$2::uuid),
  (SELECT count(*) FROM gateway_identity_pairing_start_reservation WHERE attempt_id=$2::uuid AND state='starting')`,
		draft.ID, attemptID).Scan(&activeRevisions, &draftRevisions, &exchanges, &startingReservations); err != nil {
		t.Fatalf("read pairing commit lifecycle invariant: %v", err)
	}
	if activeRevisions != 1 || draftRevisions != 0 || exchanges != 0 || startingReservations != 1 {
		t.Fatalf("pairing commit race left active=%d draft=%d exchanges=%d starting reservations=%d; "+
			"want active=1 draft=0 exchanges=0 reservation=1",
			activeRevisions, draftRevisions, exchanges, startingReservations)
	}
}

func TestConcurrentIdentityDisableAndActivationPreserveSingleActiveRevision(t *testing.T) {
	db := newIdentityPairingPostgresTestStore(t)
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	seedIdentityActivationPrerequisites(t, ctx, db, "default")
	previousID := seedIdentityLifecycleRevision(t, ctx, db, identity.RevisionActive, "default")
	nextID := seedIdentityLifecycleRevision(t, ctx, db, identity.RevisionValidated, "default")

	gate, err := db.pool.Begin(ctx)
	if err != nil {
		t.Fatalf("begin identity lifecycle gate: %v", err)
	}
	defer gate.Rollback(context.Background())
	if err := lockIdentityConfigurationLifecycle(ctx, gate); err != nil {
		t.Fatalf("lock identity lifecycle gate: %v", err)
	}

	disableResult := make(chan error, 1)
	go func() {
		_, disableErr := db.DisableActiveIdentityRevision(ctx, 1, "trace-disable", "audit-disable")
		disableResult <- disableErr
	}()
	waitForIdentityLifecycleLockWaiters(t, ctx, db, 1)

	activateResult := make(chan error, 1)
	go func() {
		_, _, activateErr := db.ActivateIdentityRevision(ctx, nextID, 1, "trace-activate", "audit-activate")
		activateResult <- activateErr
	}()
	waitForIdentityLifecycleLockWaiters(t, ctx, db, 2)
	if err := gate.Commit(ctx); err != nil {
		t.Fatalf("release identity lifecycle gate: %v", err)
	}

	if err := <-disableResult; err != nil {
		t.Fatalf("disable first queued active revision: %v", err)
	}
	activateErr := <-activateResult
	if activateErr != nil {
		t.Fatalf("activate revision after first queued disable: %v", activateErr)
	}

	var activeIDs []string
	rows, err := db.pool.Query(ctx, `SELECT id::text FROM gateway_identity_configuration_revisions
WHERE state='active' ORDER BY id`)
	if err != nil {
		t.Fatalf("read active revisions after disable/activate race: %v", err)
	}
	defer rows.Close()
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			t.Fatalf("scan active revision after disable/activate race: %v", err)
		}
		activeIDs = append(activeIDs, id)
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("iterate active revisions after disable/activate race: %v", err)
	}
	if len(activeIDs) != 1 || activeIDs[0] != nextID {
		t.Fatalf("disable/activate race left active revisions=%v; previous=%q next=%q",
			activeIDs, previousID, nextID)
	}
}

func TestCompletedIdentityPairingRestoreUsesCanonicalVisibleRevision(t *testing.T) {
	db := newIdentityPairingPostgresTestStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	var latestRevisionID string

	for index, createdAt := range []time.Time{now.Add(-time.Minute), now} {
		revisionID, pairingID := uuid.NewString(), uuid.NewString()
		if _, err := db.pool.Exec(ctx, `
INSERT INTO gateway_identity_configuration_revisions (
  id,state,auth_center_url,local_tenant_key,public_base_url,web_base_url,created_at,updated_at
) VALUES ($1::uuid,'validated','https://auth.test.example','default','https://gateway.test.example',
  'https://gateway-web.test.example',$2,$2)`, revisionID, createdAt); err != nil {
			t.Fatalf("seed completed revision %d: %v", index, err)
		}
		if _, err := db.CreateIdentityPairingExchange(ctx, identity.PairingExchange{
			ID: pairingID, RevisionID: revisionID, RemoteExchangeID: uuid.NewString(),
			ExchangeTokenRef: "identity-exchange-" + pairingID, Status: identity.PairingCompleted,
			CleanupStatus: identity.PairingCleanupNone, RemoteVersion: 4, ExpiresAt: now.Add(time.Hour),
		}); err != nil {
			t.Fatalf("seed completed pairing %d: %v", index, err)
		}
		latestRevisionID = revisionID
	}

	completed, err := db.CompletedIdentityPairingsAwaitingActivation(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(completed) != 1 || completed[0].RevisionID != latestRevisionID {
		t.Fatalf("restored completed pairings=%#v, want only canonical latest revision %s", completed, latestRevisionID)
	}
	visible, err := db.LatestInactiveIdentityConfigurationRevision(ctx)
	if err != nil || visible.ID != latestRevisionID {
		t.Fatalf("visible inactive revision=%#v error=%v, want same canonical revision", visible, err)
	}
}

func TestIdentityPairingPersistsBindingErrorCategoryAllowedByLatestMigration(t *testing.T) {
	db := newIdentityPairingPostgresTestStore(t)
	ctx := context.Background()
	revisionID := uuid.NewString()
	if _, err := db.pool.Exec(ctx, `INSERT INTO gateway_identity_configuration_revisions (
  id,state,auth_center_url,local_tenant_key,public_base_url,web_base_url
) VALUES ($1::uuid,'draft','https://auth.test.example','default','https://gateway.test.example',
  'https://gateway-web.test.example')`, revisionID); err != nil {
		t.Fatalf("seed revision for binding error category: %v", err)
	}
	pairingID := uuid.NewString()
	created, err := db.CreateIdentityPairingExchange(ctx, identity.PairingExchange{
		ID: pairingID, RevisionID: revisionID, RemoteExchangeID: uuid.NewString(),
		ExchangeTokenRef: "identity-exchange-" + pairingID, Status: identity.PairingCredentialsSaved,
		CleanupStatus: identity.PairingCleanupNone, RemoteVersion: 4, ExpiresAt: time.Now().Add(time.Hour),
	})
	if err != nil {
		t.Fatalf("create pairing for binding error category: %v", err)
	}

	const category = "security_event_connection_binding_mismatch"
	updated, err := db.RecordIdentityPairingRetryFailure(ctx, created.ID, created.Status, category)
	if err != nil {
		t.Fatalf("persist binding error category: %v", err)
	}
	if updated.LastErrorCategory != category {
		t.Fatalf("binding error category=%q, want %q", updated.LastErrorCategory, category)
	}
	var persisted string
	if err := db.pool.QueryRow(ctx, `SELECT last_error_category FROM gateway_identity_onboarding_exchanges
WHERE id=$1::uuid`, created.ID).Scan(&persisted); err != nil {
		t.Fatalf("read persisted binding error category: %v", err)
	}
	if persisted != category {
		t.Fatalf("persisted binding error category=%q, want %q", persisted, category)
	}
}

func assertIdentityRevisionSecretRefs(t *testing.T, ctx context.Context, db *Store, revisionID, wantMachine, wantSession string) {
	t.Helper()
	revision, err := db.IdentityConfigurationRevision(ctx, revisionID)
	if err != nil {
		t.Fatalf("read revision %s: %v", revisionID, err)
	}
	if revision.MachineCredentialRef != wantMachine || revision.SessionEncryptionKeyRef != wantSession {
		t.Fatalf("revision %s secret refs machine=%q session=%q, want machine=%q session=%q",
			revisionID, revision.MachineCredentialRef, revision.SessionEncryptionKeyRef, wantMachine, wantSession)
	}
}

func seedIdentityActivationPrerequisites(t *testing.T, ctx context.Context, db *Store, tenantKey string) {
	t.Helper()
	if _, err := db.pool.Exec(ctx, `INSERT INTO gateway_tenants(tenant_key,name,status)
VALUES($1,$2,'active')`, tenantKey, "Identity lifecycle test tenant"); err != nil {
		t.Fatalf("seed identity lifecycle tenant: %v", err)
	}
	if _, err := db.pool.Exec(ctx, `INSERT INTO gateway_users(user_key,username,password_hash,roles,source,status)
VALUES($1,$2,$3,'["manager"]'::jsonb,'gateway','active')`,
		"identity-lifecycle-manager", "identity-lifecycle-manager", "test-only-password-hash"); err != nil {
		t.Fatalf("seed identity lifecycle break-glass manager: %v", err)
	}
}

func seedIdentityLifecycleRevision(
	t *testing.T,
	ctx context.Context,
	db *Store,
	state identity.RevisionState,
	tenantKey string,
) string {
	t.Helper()
	revisionID := uuid.NewString()
	if _, err := db.pool.Exec(ctx, `INSERT INTO gateway_identity_configuration_revisions (
  id,state,auth_center_url,issuer,tenant_id,application_id,audience,browser_client_id,
  local_tenant_key,public_base_url,web_base_url,validated_at,activated_at
) VALUES (
  $1::uuid,$2,'https://auth.test.example','https://issuer.test.example',$3,$4,$5,$6,
  $7,'https://gateway.test.example','https://gateway-web.test.example',
  CASE WHEN $2 IN ('validated','active') THEN now() ELSE NULL END,
  CASE WHEN $2='active' THEN now() ELSE NULL END
)`, revisionID, state, uuid.NewString(), uuid.NewString(), "urn:easyai:resource:"+revisionID,
		"browser-"+revisionID, tenantKey); err != nil {
		t.Fatalf("seed %s identity lifecycle revision: %v", state, err)
	}
	return revisionID
}

func waitForIdentityLifecycleLockWaiters(t *testing.T, ctx context.Context, db *Store, want int) {
	t.Helper()
	classID := int64(uint64(identityConfigurationLifecycleLockID) >> 32)
	objectID := int64(uint64(identityConfigurationLifecycleLockID) & 0xffffffff)
	deadline := time.Now().Add(5 * time.Second)
	for {
		var waiters int
		if err := db.pool.QueryRow(ctx, `SELECT count(*) FROM pg_locks
WHERE locktype='advisory' AND database=(SELECT oid FROM pg_database WHERE datname=current_database())
  AND classid::bigint=$1 AND objid::bigint=$2 AND objsubid=1 AND NOT granted`, classID, objectID).Scan(&waiters); err != nil {
			t.Fatalf("read identity lifecycle lock waiters: %v", err)
		}
		if waiters >= want {
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf("identity lifecycle lock waiters=%d, want at least %d", waiters, want)
		}
		time.Sleep(10 * time.Millisecond)
	}
}

func newIdentityPairingPostgresTestStore(t *testing.T) *Store {
	t.Helper()
	databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL"))
	if databaseURL == "" {
		t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run identity pairing PostgreSQL integration tests")
	}
	ctx := context.Background()
	admin, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect identity pairing test database: %v", err)
	}
	var databaseName string
	if err := admin.QueryRow(ctx, `SELECT current_database()`).Scan(&databaseName); err != nil {
		admin.Close()
		t.Fatalf("read identity pairing test database name: %v", err)
	}
	if !strings.Contains(strings.ToLower(databaseName), "test") {
		admin.Close()
		t.Fatalf("refusing to use non-test database %q", databaseName)
	}

	schemaName := "gateway_identity_pairing_" + strings.ReplaceAll(uuid.NewString(), "-", "")
	schemaIdentifier := pgx.Identifier{schemaName}.Sanitize()
	if _, err := admin.Exec(ctx, `CREATE SCHEMA `+schemaIdentifier); err != nil {
		admin.Close()
		t.Fatalf("create identity pairing test schema: %v", err)
	}
	config, err := pgxpool.ParseConfig(databaseURL)
	if err != nil {
		_, _ = admin.Exec(ctx, `DROP SCHEMA IF EXISTS `+schemaIdentifier+` CASCADE`)
		admin.Close()
		t.Fatalf("parse identity pairing test database URL: %v", err)
	}
	config.ConnConfig.RuntimeParams["search_path"] = schemaName
	pool, err := pgxpool.NewWithConfig(ctx, config)
	if err != nil {
		_, _ = admin.Exec(ctx, `DROP SCHEMA IF EXISTS `+schemaIdentifier+` CASCADE`)
		admin.Close()
		t.Fatalf("connect identity pairing test schema: %v", err)
	}
	t.Cleanup(func() {
		pool.Close()
		if _, err := admin.Exec(context.Background(), `DROP SCHEMA IF EXISTS `+schemaIdentifier+` CASCADE`); err != nil {
			t.Errorf("drop identity pairing test schema: %v", err)
		}
		admin.Close()
	})

	_, filename, _, ok := runtime.Caller(0)
	if !ok {
		t.Fatal("locate identity pairing integration test")
	}
	migrationDirectory := filepath.Join(filepath.Dir(filename), "..", "..", "migrations")
	for _, migrationName := range []string{
		"0001_init.sql",
		"0061_oidc_server_sessions.sql",
		"0065_identity_configuration_revisions.sql",
		"0066_identity_onboarding_exchanges.sql",
		"0067_identity_secret_cleanup_queue.sql",
		"0068_identity_pairing_start_reservation.sql",
	} {
		migration, err := os.ReadFile(filepath.Join(migrationDirectory, migrationName))
		if err != nil {
			t.Fatalf("read %s: %v", migrationName, err)
		}
		tx, err := pool.Begin(ctx)
		if err != nil {
			t.Fatalf("begin %s: %v", migrationName, err)
		}
		if _, err := tx.Exec(ctx, string(migration)); err != nil {
			_ = tx.Rollback(ctx)
			t.Fatalf("execute %s: %v", migrationName, err)
		}
		if err := tx.Commit(ctx); err != nil {
			t.Fatalf("commit %s: %v", migrationName, err)
		}
	}
	return &Store{pool: pool}
}