package connector_oauth

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
)

type authorizationRecord struct {
	ID                        string
	StateDigest               string
	UserID                    string
	ConnectorKey              string
	ExpectedGeneration        int64
	RedirectURI               string
	CodeVerifierCiphertext    string
	RequestedScopes           string
	ExpiresAt                 time.Time
	ConsumedAt                *time.Time
	OAuthEffectState          string
	OAuthEffectVersion        int64
	OAuthActionOwnerID        string
	OAuthActionEpoch          int64
	OAuthActionLeaseExpiresAt *time.Time
	OAuthExchangeDispatchedAt *time.Time
	PendingTokenCiphertext    string
	CreatedDate               time.Time
	LastModifiedDate          time.Time
}

type tokenRecord struct {
	ConnectedPrincipalID     string
	PrincipalGeneration      int64
	AccessTokenCiphertext    string
	RefreshTokenCiphertext   string
	TokenType                string
	GrantedScopes            string
	ExpiresAt                *time.Time
	EncryptionKeyID          string
	TokenVersion             int64
	RefreshOwnershipEpoch    int64
	RefreshOwnerID           string
	RefreshLeaseExpiresAt    *time.Time
	RefreshDispatchStartedAt *time.Time
	OAuthEffectState         string
	OAuthEffectVersion       int64
	PendingTokenCiphertext   string
	CreatedDate              time.Time
	LastModifiedDate         time.Time
}

type tokenMaterial struct {
	AccessCiphertext  string
	RefreshCiphertext string
	TokenType         string
	GrantedScopes     string
	ExpiresAt         *time.Time
	EncryptionKeyID   string
}

type refreshClaim struct {
	Principal ConnectedPrincipal
	Token     tokenRecord
	OwnerID   string
	Epoch     int64
}

type authorizationClaim struct {
	Record  authorizationRecord
	OwnerID string
	Epoch   int64
}

type revokeClaim struct {
	Principal ConnectedPrincipal
	Token     tokenRecord
	OwnerID   string
	Epoch     int64
}

func validateAuthorizationEffectShape(record authorizationRecord) error {
	switch record.OAuthEffectState {
	case EffectNone:
		if record.ConsumedAt != nil || record.OAuthActionOwnerID != "" || record.OAuthActionLeaseExpiresAt != nil ||
			record.OAuthExchangeDispatchedAt != nil || record.PendingTokenCiphertext != "" {
			return ErrEffectStateInvalid
		}
	case EffectExchangeClaimed:
		if record.ConsumedAt == nil || record.OAuthActionOwnerID == "" || record.OAuthActionLeaseExpiresAt == nil ||
			record.OAuthExchangeDispatchedAt != nil || record.PendingTokenCiphertext != "" {
			return ErrEffectStateInvalid
		}
	case EffectExchangeDispatched:
		if record.ConsumedAt == nil || record.OAuthActionOwnerID == "" || record.OAuthActionLeaseExpiresAt == nil ||
			record.OAuthExchangeDispatchedAt == nil || record.PendingTokenCiphertext != "" {
			return ErrEffectStateInvalid
		}
	case EffectExchangeResultStaged:
		if record.ConsumedAt == nil || record.OAuthActionOwnerID != "" || record.OAuthActionLeaseExpiresAt != nil ||
			record.OAuthExchangeDispatchedAt == nil || record.PendingTokenCiphertext == "" {
			return ErrEffectStateInvalid
		}
	case EffectManualRequired:
		if record.ConsumedAt == nil || record.OAuthActionOwnerID != "" || record.OAuthActionLeaseExpiresAt != nil {
			return ErrEffectStateInvalid
		}
	default:
		return ErrEffectStateInvalid
	}
	return nil
}

func validateTokenEffectShape(token tokenRecord) error {
	switch token.OAuthEffectState {
	case EffectNone, EffectRevokePending:
		if token.RefreshOwnerID != "" || token.RefreshLeaseExpiresAt != nil ||
			token.RefreshDispatchStartedAt != nil || token.PendingTokenCiphertext != "" {
			return ErrEffectStateInvalid
		}
	case EffectRefreshClaimed, EffectRevokeClaimed:
		if token.RefreshOwnerID == "" || token.RefreshLeaseExpiresAt == nil ||
			token.RefreshDispatchStartedAt != nil || token.PendingTokenCiphertext != "" {
			return ErrEffectStateInvalid
		}
	case EffectRefreshDispatched, EffectRevokeDispatched:
		if token.RefreshOwnerID == "" || token.RefreshLeaseExpiresAt == nil ||
			token.RefreshDispatchStartedAt == nil || token.PendingTokenCiphertext != "" {
			return ErrEffectStateInvalid
		}
	case EffectRefreshResultStaged:
		if token.RefreshOwnerID != "" || token.RefreshLeaseExpiresAt != nil ||
			token.RefreshDispatchStartedAt == nil || token.PendingTokenCiphertext == "" {
			return ErrEffectStateInvalid
		}
	case EffectRevokeSucceeded:
		if token.RefreshOwnerID != "" || token.RefreshLeaseExpiresAt != nil ||
			token.RefreshDispatchStartedAt == nil || token.PendingTokenCiphertext != "" {
			return ErrEffectStateInvalid
		}
	case EffectManualRequired:
		if token.RefreshOwnerID != "" || token.RefreshLeaseExpiresAt != nil {
			return ErrEffectStateInvalid
		}
	default:
		return ErrEffectStateInvalid
	}
	return nil
}

var (
	errAuthorizationReservePostCommitCleanup  = errors.New("connected OAuth authorization reservation post-commit cleanup failed")
	errAuthorizationReserveUnconfirmedCleanup = errors.New("connected OAuth authorization reservation unconfirmed cleanup failed")
	errAuthorizationReleasePostCommitCleanup  = errors.New("connected OAuth authorization release post-commit cleanup failed")
	errAuthorizationReleaseUnconfirmedCleanup = errors.New("connected OAuth authorization release unconfirmed cleanup failed")
	errDisconnectPostCommitCleanup            = errors.New("connected OAuth disconnect post-commit cleanup failed")
	errDisconnectUnconfirmedCleanup           = errors.New("connected OAuth disconnect unconfirmed cleanup failed")
	errOAuthRepositoryTransactionCleanup      = errors.New("connected OAuth repository transaction cleanup failed")
)

type authorizationReservationResult struct {
	record                  *authorizationRecord
	localCommitConfirmed    bool
	postCommitCleanupFailed error
}

type authorizationReleaseResult struct {
	deleteCommitConfirmed   bool
	postCommitCleanupFailed error
}

type disconnectResult struct {
	principal               *ConnectedPrincipal
	token                   *tokenRecord
	localCommitConfirmed    bool
	postCommitCleanupFailed error
}

type oauthRepository interface {
	reserveAuthorization(context.Context, authorizationRecord, func(int64) (string, error)) (authorizationReservationResult, error)
	releaseAuthorization(context.Context, authorizationRecord, bool) (authorizationReleaseResult, error)
	claimAuthorizationExchange(context.Context, string, string, string, time.Time, time.Time) (*authorizationClaim, error)
	markAuthorizationExchangeDispatched(context.Context, authorizationClaim, time.Time) error
	stageAuthorizationResult(context.Context, authorizationClaim, string, time.Time) error
	reconcileAuthorizationStage(context.Context, authorizationClaim, string, time.Time) (bool, error)
	markAuthorizationManual(context.Context, authorizationClaim, time.Time) error
	markStagedAuthorizationManual(context.Context, authorizationRecord, time.Time) error
	finalizeAuthorization(context.Context, authorizationRecord, RemoteIdentity, string, time.Time, func(ConnectedPrincipal) (tokenMaterial, error)) (*ConnectedPrincipal, error)
	loadPrincipal(context.Context, string, string) (*ConnectedPrincipal, error)
	resolveToken(context.Context, string, string, int64, time.Time) (*ConnectedPrincipal, *tokenRecord, error)
	claimRefresh(context.Context, string, string, int64, string, time.Time, time.Time) (*refreshClaim, error)
	markRefreshDispatched(context.Context, refreshClaim, time.Time) error
	stageRefreshResult(context.Context, refreshClaim, string, time.Time) error
	reconcileRefreshStage(context.Context, refreshClaim, string, time.Time) (bool, error)
	finalizeStagedRefresh(context.Context, ConnectedPrincipal, tokenRecord, tokenMaterial, time.Time) error
	markRefreshManual(context.Context, refreshClaim, time.Time) error
	markStagedRefreshManual(context.Context, ConnectedPrincipal, tokenRecord, time.Time) error
	disconnect(context.Context, string, string, int64, string, time.Time) (disconnectResult, error)
	claimRevoke(context.Context, string, string, string, time.Time, time.Time) (*revokeClaim, error)
	markRevokeDispatched(context.Context, revokeClaim, time.Time) error
	markRevokeManual(context.Context, revokeClaim, time.Time) error
	markRevokeSucceeded(context.Context, revokeClaim, time.Time) error
	deleteRevokedToken(context.Context, string, string, string, int64, int64) error
	loadEffectStatus(context.Context, string, string, time.Time) (EffectStatus, error)
	attestExternalCleanup(context.Context, CleanupAttestation, time.Time, CleanupAuditEmitter) error
}

// SQLRepository persists connected-account OAuth state in the catalog-owned
// system tables.
type SQLRepository struct {
	db *sql.DB
}

func NewSQLRepository(db *sql.DB) (*SQLRepository, error) {
	if db == nil {
		return nil, ErrDependencyMissing
	}
	return &SQLRepository{db: db}, nil
}

func (r *SQLRepository) reserveAuthorization(ctx context.Context, record authorizationRecord, seal func(int64) (string, error)) (authorizationReservationResult, error) {
	if r == nil || r.db == nil || seal == nil {
		return authorizationReservationResult{}, ErrDependencyMissing
	}
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return authorizationReservationResult{}, err
	}
	var outcome authorizationReservationResult
	transactionErr := func() error {
		if err := lockUser(ctx, tx, record.UserID); err != nil {
			return err
		}
		existingAuthorization, err := scanAuthorization(tx.QueryRowContext(ctx, authorizationSelect+`
 WHERE UserId = ? AND ConnectorKey = ?
 LIMIT 1
 FOR UPDATE`, record.UserID, record.ConnectorKey))
		if err == nil {
			if err := validateAuthorizationEffectShape(*existingAuthorization); err != nil {
				return err
			}
			switch existingAuthorization.OAuthEffectState {
			case EffectExchangeResultStaged:
				outcome.record = existingAuthorization
				return nil
			case EffectExchangeDispatched:
				return ErrAuthorizationProgress
			case EffectManualRequired:
				return ErrManualActionRequired
			case EffectNone:
				if existingAuthorization.ExpiresAt.After(record.CreatedDate) {
					return ErrAuthorizationProgress
				}
				result, deleteErr := tx.ExecContext(ctx, `DELETE FROM _System_ConnectedAuthorizationRequest
	 WHERE Id = ? AND OAuthEffectState = ? AND OAuthEffectVersion = ?
	   AND ConsumedAt IS NULL
	   AND OAuthActionOwnerId IS NULL AND OAuthActionLeaseExpiresAt IS NULL
   AND OAuthExchangeDispatchedAt IS NULL AND PendingTokenCiphertext IS NULL`,
					existingAuthorization.ID, EffectNone, existingAuthorization.OAuthEffectVersion)
				if deleteErr != nil {
					return deleteErr
				}
				deleted, deleteErr := exactlyOneRow(result)
				if deleteErr != nil {
					return deleteErr
				}
				if !deleted {
					return ErrOwnershipLost
				}
				err = sql.ErrNoRows
			case EffectExchangeClaimed:
				if existingAuthorization.OAuthActionLeaseExpiresAt != nil && existingAuthorization.OAuthActionLeaseExpiresAt.After(record.CreatedDate) {
					return ErrAuthorizationProgress
				}
				result, deleteErr := tx.ExecContext(ctx, `DELETE FROM _System_ConnectedAuthorizationRequest
	 WHERE Id = ? AND OAuthEffectState = ? AND OAuthEffectVersion = ? AND OAuthActionEpoch = ?
	   AND ConsumedAt = ?
	   AND BINARY OAuthActionOwnerId = BINARY ? AND OAuthActionLeaseExpiresAt = ?
	   AND OAuthActionLeaseExpiresAt <= ? AND OAuthExchangeDispatchedAt IS NULL
	   AND PendingTokenCiphertext IS NULL`, existingAuthorization.ID, EffectExchangeClaimed,
					existingAuthorization.OAuthEffectVersion, existingAuthorization.OAuthActionEpoch,
					existingAuthorization.ConsumedAt,
					existingAuthorization.OAuthActionOwnerID, existingAuthorization.OAuthActionLeaseExpiresAt,
					record.CreatedDate)
				if deleteErr != nil {
					return deleteErr
				}
				deleted, deleteErr := exactlyOneRow(result)
				if deleteErr != nil {
					return deleteErr
				}
				if !deleted {
					return ErrOwnershipLost
				}
				err = sql.ErrNoRows
			default:
				return ErrEffectStateInvalid
			}
		}
		if !errors.Is(err, sql.ErrNoRows) {
			return err
		}
		principal, err := scanPrincipal(tx.QueryRowContext(ctx, principalSelect+`
 WHERE UserId = ? AND ConnectorKey = ?
 LIMIT 1
 FOR UPDATE`, record.UserID, record.ConnectorKey))
		if err != nil && !errors.Is(err, sql.ErrNoRows) {
			return err
		}
		record.ExpectedGeneration = 1
		if principal != nil {
			existingToken, tokenErr := scanToken(tx.QueryRowContext(ctx, tokenSelect+`
 WHERE ConnectedPrincipalId = ? LIMIT 1 FOR UPDATE`, principal.ID))
			if tokenErr != nil && !errors.Is(tokenErr, sql.ErrNoRows) {
				return tokenErr
			}
			if existingToken != nil {
				return ErrDisconnectRequired
			}
			record.ExpectedGeneration = principal.Generation + 1
		}
		record.CodeVerifierCiphertext, err = seal(record.ExpectedGeneration)
		if err != nil {
			return err
		}
		_, err = tx.ExecContext(ctx, `
INSERT INTO _System_ConnectedAuthorizationRequest
       (Id, StateDigest, UserId, ConnectorKey, ExpectedGeneration, RedirectUri,
        CodeVerifierCiphertext, RequestedScopes, ExpiresAt, ConsumedAt, OAuthEffectState,
        OAuthEffectVersion, OAuthActionEpoch, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, 0, 0, ?, ?)`,
			record.ID, record.StateDigest, record.UserID, record.ConnectorKey, record.ExpectedGeneration,
			record.RedirectURI, record.CodeVerifierCiphertext, record.RequestedScopes, record.ExpiresAt,
			EffectNone, record.CreatedDate, record.LastModifiedDate)
		if err == nil {
			record.OAuthEffectState = EffectNone
			reserved := record
			outcome.record = &reserved
		}
		return err
	}()
	if transactionErr != nil {
		_ = tx.Rollback()
		return authorizationReservationResult{}, joinOAuthCleanupErrors(transactionErr, cleanup(), errAuthorizationReserveUnconfirmedCleanup)
	}
	if err := tx.Commit(); err != nil {
		_ = tx.Rollback()
		return authorizationReservationResult{}, joinOAuthCleanupErrors(err, cleanup(), errAuthorizationReserveUnconfirmedCleanup)
	}
	outcome.localCommitConfirmed = true
	if err := cleanup(); err != nil {
		outcome.postCommitCleanupFailed = errAuthorizationReservePostCommitCleanup
	}
	return outcome, nil
}

func (r *SQLRepository) releaseAuthorization(ctx context.Context, record authorizationRecord, consumed bool) (authorizationReleaseResult, error) {
	if r == nil || r.db == nil {
		return authorizationReleaseResult{}, ErrDependencyMissing
	}
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return authorizationReleaseResult{}, err
	}
	transactionErr := func() error {
		if err := lockUser(ctx, tx, record.UserID); err != nil {
			return err
		}
		consumedPredicate := "ConsumedAt IS NULL"
		if consumed {
			consumedPredicate = "ConsumedAt IS NOT NULL"
		}
		_, err := tx.ExecContext(ctx, `
DELETE FROM _System_ConnectedAuthorizationRequest
 WHERE Id = ? AND UserId = ? AND ConnectorKey = ? AND ExpectedGeneration = ?
   AND `+consumedPredicate,
			record.ID, record.UserID, record.ConnectorKey, record.ExpectedGeneration)
		return err
	}()
	if transactionErr != nil {
		_ = tx.Rollback()
		return authorizationReleaseResult{}, joinOAuthCleanupErrors(transactionErr, cleanup(), errAuthorizationReleaseUnconfirmedCleanup)
	}
	if err := tx.Commit(); err != nil {
		_ = tx.Rollback()
		return authorizationReleaseResult{}, joinOAuthCleanupErrors(err, cleanup(), errAuthorizationReleaseUnconfirmedCleanup)
	}
	outcome := authorizationReleaseResult{deleteCommitConfirmed: true}
	if err := cleanup(); err != nil {
		outcome.postCommitCleanupFailed = errAuthorizationReleasePostCommitCleanup
	}
	return outcome, nil
}

func (r *SQLRepository) finalizeAuthorization(ctx context.Context, record authorizationRecord, identity RemoteIdentity, candidateID string, now time.Time, seal func(ConnectedPrincipal) (tokenMaterial, error)) (*ConnectedPrincipal, error) {
	if r == nil || r.db == nil || seal == nil {
		return nil, ErrDependencyMissing
	}
	var finalized ConnectedPrincipal
	err := withPessimisticTx(ctx, r.db, func(tx *sql.Tx) error {
		if err := lockUser(ctx, tx, record.UserID); err != nil {
			return err
		}
		lockedAuth, err := scanAuthorization(tx.QueryRowContext(ctx, authorizationSelect+`
 WHERE Id = ? AND UserId = ?
 LIMIT 1
 FOR UPDATE`, record.ID, record.UserID))
		if err != nil {
			return err
		}
		if err := validateAuthorizationEffectShape(*lockedAuth); err != nil {
			return err
		}
		if lockedAuth.ConsumedAt == nil || lockedAuth.ExpectedGeneration != record.ExpectedGeneration ||
			lockedAuth.OAuthEffectState != EffectExchangeResultStaged || lockedAuth.PendingTokenCiphertext == "" {
			return ErrAuthorizationDenied
		}
		existing, err := scanPrincipal(tx.QueryRowContext(ctx, principalSelect+`
 WHERE UserId = ? AND ConnectorKey = ?
 LIMIT 1
 FOR UPDATE`, record.UserID, record.ConnectorKey))
		if err != nil && !errors.Is(err, sql.ErrNoRows) {
			return err
		}
		currentGeneration := int64(0)
		if existing != nil {
			currentGeneration = existing.Generation
		}
		if currentGeneration != record.ExpectedGeneration-1 {
			return ErrGenerationMismatch
		}
		remote, lookupErr := scanPrincipal(tx.QueryRowContext(ctx, principalSelect+`
 WHERE ConnectorKey = ? AND Issuer = ? AND Subject = ?
 LIMIT 1
 FOR UPDATE`, record.ConnectorKey, identity.Issuer, identity.Subject))
		if lookupErr != nil && !errors.Is(lookupErr, sql.ErrNoRows) {
			return lookupErr
		}
		if remote != nil && remote.UserID != record.UserID {
			return ErrPrincipalOwnership
		}
		principal := ConnectedPrincipal{
			ID: candidateID, UserID: record.UserID, ConnectorKey: record.ConnectorKey,
			Generation: record.ExpectedGeneration, Issuer: identity.Issuer, Subject: identity.Subject,
			PrincipalAddress: identity.PrincipalAddress, DisplayName: identity.DisplayName,
			State: PrincipalConnected, ConnectedAt: now, CreatedDate: now, LastModifiedDate: now,
		}
		if existing == nil {
			_, err = tx.ExecContext(ctx, `
INSERT INTO _System_ConnectedPrincipal
       (Id, UserId, ConnectorKey, Generation, Issuer, Subject, PrincipalAddress, DisplayName,
        State, ConnectedAt, DisconnectedAt, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?)`,
				principal.ID, principal.UserID, principal.ConnectorKey, principal.Generation, principal.Issuer,
				principal.Subject, principal.PrincipalAddress, nullableString(principal.DisplayName),
				principal.State, principal.ConnectedAt, principal.CreatedDate, principal.LastModifiedDate)
		} else {
			principal.ID = existing.ID
			principal.CreatedDate = existing.CreatedDate
			result, updateErr := tx.ExecContext(ctx, `
UPDATE _System_ConnectedPrincipal
   SET Generation = ?, Issuer = ?, Subject = ?, PrincipalAddress = ?, DisplayName = ?, State = ?,
       ConnectedAt = ?, DisconnectedAt = NULL, LastModifiedDate = ?
 WHERE Id = ? AND UserId = ? AND ConnectorKey = ? AND Generation = ?`,
				principal.Generation, principal.Issuer, principal.Subject, principal.PrincipalAddress,
				nullableString(principal.DisplayName), principal.State, principal.ConnectedAt,
				principal.LastModifiedDate, principal.ID, principal.UserID, principal.ConnectorKey, currentGeneration)
			if updateErr != nil {
				return classifyPrincipalWriteError(updateErr)
			}
			won, updateErr := exactlyOneRow(result)
			if updateErr != nil || !won {
				if updateErr != nil {
					return updateErr
				}
				return ErrGenerationMismatch
			}
		}
		if err != nil {
			return classifyPrincipalWriteError(err)
		}
		material, err := seal(principal)
		if err != nil {
			return err
		}
		currentToken, err := scanToken(tx.QueryRowContext(ctx, tokenSelect+`
 WHERE ConnectedPrincipalId = ?
 LIMIT 1
 FOR UPDATE`, principal.ID))
		if err != nil && !errors.Is(err, sql.ErrNoRows) {
			return err
		}
		if currentToken != nil {
			return ErrDisconnectRequired
		}
		tokenVersion := int64(1)
		createdDate := now
		if currentToken != nil {
			tokenVersion = currentToken.TokenVersion + 1
			createdDate = currentToken.CreatedDate
			result, updateErr := tx.ExecContext(ctx, `
UPDATE _System_ConnectedPrincipalToken
   SET PrincipalGeneration = ?, AccessTokenCiphertext = ?, RefreshTokenCiphertext = ?, TokenType = ?,
       GrantedScopes = ?, ExpiresAt = ?, EncryptionKeyId = ?, TokenVersion = ?,
				RefreshOwnershipEpoch = RefreshOwnershipEpoch + 1, RefreshOwnerId = NULL,
				RefreshLeaseExpiresAt = NULL, RefreshDispatchStartedAt = NULL,
				OAuthEffectState = ?, OAuthEffectVersion = OAuthEffectVersion + 1,
				PendingTokenCiphertext = NULL, LastModifiedDate = ?
 WHERE ConnectedPrincipalId = ? AND TokenVersion = ?`,
				principal.Generation, material.AccessCiphertext, nullableString(material.RefreshCiphertext),
				material.TokenType, material.GrantedScopes, nullableTime(material.ExpiresAt), material.EncryptionKeyID,
				tokenVersion, EffectNone, now, principal.ID, currentToken.TokenVersion)
			if updateErr != nil {
				return updateErr
			}
			won, updateErr := exactlyOneRow(result)
			if updateErr != nil {
				return updateErr
			}
			if !won {
				return ErrGenerationMismatch
			}
		} else {
			_, err = tx.ExecContext(ctx, `
INSERT INTO _System_ConnectedPrincipalToken
	       (ConnectedPrincipalId, PrincipalGeneration, AccessTokenCiphertext, RefreshTokenCiphertext,
	        TokenType, GrantedScopes, ExpiresAt, EncryptionKeyId, TokenVersion, RefreshOwnershipEpoch,
	        RefreshOwnerId, RefreshLeaseExpiresAt, RefreshDispatchStartedAt, OAuthEffectState,
	        OAuthEffectVersion, PendingTokenCiphertext, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0, NULL, NULL, NULL, ?, 0, NULL, ?, ?)`,
				principal.ID, principal.Generation, material.AccessCiphertext, nullableString(material.RefreshCiphertext),
				material.TokenType, material.GrantedScopes, nullableTime(material.ExpiresAt), material.EncryptionKeyID,
				tokenVersion, EffectNone, createdDate, now)
		}
		if err != nil {
			return err
		}
		result, err := tx.ExecContext(ctx, `
DELETE FROM _System_ConnectedAuthorizationRequest
 WHERE Id = ? AND UserId = ? AND ExpectedGeneration = ? AND ConsumedAt IS NOT NULL`,
			record.ID, record.UserID, record.ExpectedGeneration)
		if err != nil {
			return err
		}
		deleted, err := exactlyOneRow(result)
		if err != nil {
			return err
		}
		if !deleted {
			return ErrAuthorizationDenied
		}
		finalized = principal
		return nil
	})
	if err != nil {
		return nil, err
	}
	return &finalized, nil
}

func (r *SQLRepository) loadPrincipal(ctx context.Context, userID, connectorKey string) (*ConnectedPrincipal, error) {
	if r == nil || r.db == nil {
		return nil, ErrDependencyMissing
	}
	principal, err := scanPrincipal(r.db.QueryRowContext(ctx, principalSelect+` WHERE UserId = ? AND ConnectorKey = ? LIMIT 1`, userID, connectorKey))
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return principal, err
}

func (r *SQLRepository) resolveToken(ctx context.Context, userID, connectorKey string, expectedGeneration int64, now time.Time) (*ConnectedPrincipal, *tokenRecord, error) {
	if r == nil || r.db == nil {
		return nil, nil, ErrDependencyMissing
	}
	var principal *ConnectedPrincipal
	var token *tokenRecord
	err := withPessimisticTx(ctx, r.db, func(tx *sql.Tx) error {
		if err := lockUser(ctx, tx, userID); err != nil {
			return err
		}
		var err error
		principal, err = scanPrincipal(tx.QueryRowContext(ctx, principalSelect+`
 WHERE UserId = ? AND ConnectorKey = ?
 LIMIT 1
 FOR UPDATE`, userID, connectorKey))
		if errors.Is(err, sql.ErrNoRows) {
			return ErrPrincipalNotFound
		}
		if err != nil {
			return err
		}
		if principal.State != PrincipalConnected {
			return ErrReconnectRequired
		}
		if principal.Generation != expectedGeneration {
			return ErrGenerationMismatch
		}
		token, err = scanToken(tx.QueryRowContext(ctx, tokenSelect+`
 WHERE ConnectedPrincipalId = ?
 LIMIT 1
 FOR UPDATE`, principal.ID))
		if errors.Is(err, sql.ErrNoRows) {
			return ErrReconnectRequired
		}
		if err != nil {
			return err
		}
		if token.PrincipalGeneration != expectedGeneration {
			return ErrReconnectRequired
		}
		if err := validateTokenEffectShape(*token); err != nil {
			return err
		}
		switch token.OAuthEffectState {
		case EffectNone:
			if token.RefreshOwnerID != "" || token.RefreshLeaseExpiresAt != nil ||
				token.RefreshDispatchStartedAt != nil || token.PendingTokenCiphertext != "" {
				return ErrEffectStateInvalid
			}
		case EffectRefreshResultStaged:
			if token.RefreshOwnerID != "" || token.RefreshLeaseExpiresAt != nil ||
				token.RefreshDispatchStartedAt == nil || token.PendingTokenCiphertext == "" {
				return ErrEffectStateInvalid
			}
		case EffectRefreshClaimed:
			if token.RefreshDispatchStartedAt != nil || token.RefreshOwnerID == "" ||
				token.RefreshLeaseExpiresAt == nil || token.PendingTokenCiphertext != "" {
				return ErrEffectStateInvalid
			}
			if token.RefreshLeaseExpiresAt.After(now) {
				return ErrRefreshInProgress
			}
			result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectedPrincipalToken
   SET OAuthEffectState = ?, OAuthEffectVersion = OAuthEffectVersion + 1,
       RefreshOwnerId = NULL, RefreshLeaseExpiresAt = NULL, LastModifiedDate = ?
 WHERE ConnectedPrincipalId = ? AND PrincipalGeneration = ? AND TokenVersion = ?
   AND RefreshOwnershipEpoch = ? AND RefreshOwnerId = ? AND OAuthEffectVersion = ?
   AND OAuthEffectState = ? AND RefreshDispatchStartedAt IS NULL
	   AND RefreshLeaseExpiresAt IS NOT NULL AND RefreshLeaseExpiresAt <= ?
	   AND PendingTokenCiphertext IS NULL`,
				EffectNone, now, principal.ID, expectedGeneration, token.TokenVersion,
				token.RefreshOwnershipEpoch, token.RefreshOwnerID, token.OAuthEffectVersion,
				EffectRefreshClaimed, now)
			if err != nil {
				return err
			}
			won, err := exactlyOneRow(result)
			if err != nil {
				return err
			}
			if !won {
				return ErrOwnershipLost
			}
			token.OAuthEffectState = EffectNone
			token.OAuthEffectVersion++
			token.RefreshOwnerID = ""
			token.RefreshLeaseExpiresAt = nil
		case EffectRefreshDispatched:
			if token.RefreshDispatchStartedAt == nil || token.RefreshOwnerID == "" ||
				token.RefreshLeaseExpiresAt == nil || token.PendingTokenCiphertext != "" {
				return ErrEffectStateInvalid
			}
			return ErrRefreshInProgress
		case EffectRevokePending, EffectRevokeClaimed, EffectRevokeDispatched,
			EffectRevokeSucceeded, EffectManualRequired:
			return ErrReconnectRequired
		default:
			return ErrEffectStateInvalid
		}
		return nil
	})
	return principal, token, err
}

func (r *SQLRepository) claimRefresh(ctx context.Context, userID, connectorKey string, expectedGeneration int64, ownerID string, now, leaseUntil time.Time) (*refreshClaim, error) {
	if r == nil || r.db == nil {
		return nil, ErrDependencyMissing
	}
	var claim refreshClaim
	err := withPessimisticTx(ctx, r.db, func(tx *sql.Tx) error {
		if err := lockUser(ctx, tx, userID); err != nil {
			return err
		}
		principal, err := scanPrincipal(tx.QueryRowContext(ctx, principalSelect+`
 WHERE UserId = ? AND ConnectorKey = ? LIMIT 1 FOR UPDATE`, userID, connectorKey))
		if errors.Is(err, sql.ErrNoRows) {
			return ErrPrincipalNotFound
		}
		if err != nil {
			return err
		}
		if principal.State != PrincipalConnected {
			return ErrReconnectRequired
		}
		if principal.Generation != expectedGeneration {
			return ErrGenerationMismatch
		}
		token, err := scanToken(tx.QueryRowContext(ctx, tokenSelect+`
 WHERE ConnectedPrincipalId = ? LIMIT 1 FOR UPDATE`, principal.ID))
		if errors.Is(err, sql.ErrNoRows) {
			return ErrReconnectRequired
		}
		if err != nil {
			return err
		}
		if token.PrincipalGeneration != expectedGeneration || token.RefreshTokenCiphertext == "" {
			return ErrReconnectRequired
		}
		if err := validateTokenEffectShape(*token); err != nil {
			return err
		}
		switch token.OAuthEffectState {
		case EffectRefreshResultStaged:
			if token.PendingTokenCiphertext == "" {
				return ErrEffectStateInvalid
			}
			claim = refreshClaim{Principal: *principal, Token: *token}
			return nil
		case EffectRefreshDispatched:
			if token.RefreshDispatchStartedAt == nil || token.RefreshOwnerID == "" ||
				token.RefreshLeaseExpiresAt == nil || token.PendingTokenCiphertext != "" {
				return ErrEffectStateInvalid
			}
			return ErrRefreshInProgress
		case EffectRefreshClaimed:
			if token.RefreshDispatchStartedAt != nil || token.RefreshOwnerID == "" || token.RefreshLeaseExpiresAt == nil || token.PendingTokenCiphertext != "" {
				return ErrEffectStateInvalid
			}
			if token.RefreshLeaseExpiresAt.After(now) {
				return ErrRefreshInProgress
			}
		case EffectNone:
			if token.RefreshOwnerID != "" || token.RefreshLeaseExpiresAt != nil || token.RefreshDispatchStartedAt != nil || token.PendingTokenCiphertext != "" {
				return ErrEffectStateInvalid
			}
		case EffectRevokePending, EffectRevokeClaimed, EffectRevokeDispatched, EffectRevokeSucceeded, EffectManualRequired:
			return ErrReconnectRequired
		default:
			return ErrEffectStateInvalid
		}
		newEpoch := token.RefreshOwnershipEpoch + 1
		var result sql.Result
		switch token.OAuthEffectState {
		case EffectNone:
			result, err = tx.ExecContext(ctx, `
UPDATE _System_ConnectedPrincipalToken
   SET RefreshOwnershipEpoch = ?, RefreshOwnerId = ?, RefreshLeaseExpiresAt = ?,
				RefreshDispatchStartedAt = NULL, OAuthEffectState = ?,
				OAuthEffectVersion = OAuthEffectVersion + 1, PendingTokenCiphertext = NULL,
				LastModifiedDate = ?
 WHERE ConnectedPrincipalId = ? AND PrincipalGeneration = ? AND TokenVersion = ?
			   AND RefreshOwnershipEpoch = ? AND OAuthEffectVersion = ? AND OAuthEffectState = ?
			   AND RefreshOwnerId IS NULL AND RefreshLeaseExpiresAt IS NULL
			   AND RefreshDispatchStartedAt IS NULL AND PendingTokenCiphertext IS NULL`,
				newEpoch, ownerID, leaseUntil, EffectRefreshClaimed, now,
				principal.ID, expectedGeneration, token.TokenVersion, token.RefreshOwnershipEpoch,
				token.OAuthEffectVersion, EffectNone)
		case EffectRefreshClaimed:
			result, err = tx.ExecContext(ctx, `
UPDATE _System_ConnectedPrincipalToken
   SET RefreshOwnershipEpoch = ?, RefreshOwnerId = ?, RefreshLeaseExpiresAt = ?,
				RefreshDispatchStartedAt = NULL, OAuthEffectState = ?,
				OAuthEffectVersion = OAuthEffectVersion + 1, PendingTokenCiphertext = NULL,
				LastModifiedDate = ?
 WHERE ConnectedPrincipalId = ? AND PrincipalGeneration = ? AND TokenVersion = ?
			   AND RefreshOwnershipEpoch = ? AND OAuthEffectVersion = ? AND OAuthEffectState = ?
			   AND BINARY RefreshOwnerId = BINARY ? AND RefreshDispatchStartedAt IS NULL
			   AND RefreshLeaseExpiresAt IS NOT NULL AND RefreshLeaseExpiresAt <= ?
			   AND PendingTokenCiphertext IS NULL`,
				newEpoch, ownerID, leaseUntil, EffectRefreshClaimed, now,
				principal.ID, expectedGeneration, token.TokenVersion, token.RefreshOwnershipEpoch,
				token.OAuthEffectVersion, EffectRefreshClaimed, token.RefreshOwnerID, now)
		default:
			return ErrEffectStateInvalid
		}
		if err != nil {
			return err
		}
		won, err := exactlyOneRow(result)
		if err != nil {
			return err
		}
		if !won {
			return ErrOwnershipLost
		}
		token.RefreshOwnershipEpoch = newEpoch
		token.RefreshOwnerID = ownerID
		token.RefreshLeaseExpiresAt = &leaseUntil
		claim = refreshClaim{Principal: *principal, Token: *token, OwnerID: ownerID, Epoch: newEpoch}
		return nil
	})
	if err != nil {
		return nil, err
	}
	return &claim, nil
}

func (r *SQLRepository) markRefreshDispatched(ctx context.Context, claim refreshClaim, now time.Time) error {
	return r.withRefreshFence(ctx, claim, func(tx *sql.Tx) error {
		result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectedPrincipalToken
			SET RefreshDispatchStartedAt = ?, OAuthEffectState = ?,
			    OAuthEffectVersion = OAuthEffectVersion + 1, LastModifiedDate = ?
 WHERE ConnectedPrincipalId = ? AND PrincipalGeneration = ? AND TokenVersion = ?
   AND RefreshOwnershipEpoch = ? AND BINARY RefreshOwnerId = BINARY ?
			   AND RefreshDispatchStartedAt IS NULL AND OAuthEffectState = ?`, now, EffectRefreshDispatched, now,
			claim.Principal.ID, claim.Principal.Generation,
			claim.Token.TokenVersion, claim.Epoch, claim.OwnerID, EffectRefreshClaimed)
		if err != nil {
			return err
		}
		won, err := exactlyOneRow(result)
		if err != nil {
			return err
		}
		if !won {
			return ErrOwnershipLost
		}
		return nil
	})
}

func setPrincipalReconnectRequired(ctx context.Context, tx *sql.Tx, principal ConnectedPrincipal, now time.Time) error {
	result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectedPrincipal
   SET State = ?, LastModifiedDate = ?
 WHERE Id = ? AND UserId = ? AND ConnectorKey = ? AND Generation = ? AND State = ?`,
		PrincipalReconnectRequired, now, principal.ID, principal.UserID,
		principal.ConnectorKey, principal.Generation, PrincipalConnected)
	if err != nil {
		return err
	}
	won, err := exactlyOneRow(result)
	if err != nil {
		return err
	}
	if !won {
		return ErrOwnershipLost
	}
	return nil
}

func (r *SQLRepository) withRefreshFence(ctx context.Context, claim refreshClaim, fn func(*sql.Tx) error) error {
	if r == nil || r.db == nil || fn == nil {
		return ErrDependencyMissing
	}
	return withPessimisticTx(ctx, r.db, func(tx *sql.Tx) error {
		if err := lockUser(ctx, tx, claim.Principal.UserID); err != nil {
			return err
		}
		principal, err := scanPrincipal(tx.QueryRowContext(ctx, principalSelect+`
 WHERE Id = ? AND UserId = ? LIMIT 1 FOR UPDATE`, claim.Principal.ID, claim.Principal.UserID))
		if err != nil {
			return err
		}
		if principal.State != PrincipalConnected || principal.Generation != claim.Principal.Generation {
			return ErrOwnershipLost
		}
		token, err := scanToken(tx.QueryRowContext(ctx, tokenSelect+`
 WHERE ConnectedPrincipalId = ? LIMIT 1 FOR UPDATE`, claim.Principal.ID))
		if err != nil {
			return err
		}
		if err := validateTokenEffectShape(*token); err != nil {
			return err
		}
		if token.PrincipalGeneration != claim.Principal.Generation || token.TokenVersion != claim.Token.TokenVersion ||
			token.RefreshOwnershipEpoch != claim.Epoch || token.RefreshOwnerID != claim.OwnerID {
			return ErrOwnershipLost
		}
		return fn(tx)
	})
}

func (r *SQLRepository) disconnect(ctx context.Context, userID, connectorKey string, expectedGeneration int64, state string, now time.Time) (disconnectResult, error) {
	if r == nil || r.db == nil {
		return disconnectResult{}, ErrDependencyMissing
	}
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return disconnectResult{}, err
	}
	var outcome disconnectResult
	transactionErr := func() error {
		if err := lockUser(ctx, tx, userID); err != nil {
			return err
		}
		outcome.principal, err = scanPrincipal(tx.QueryRowContext(ctx, principalSelect+`
 WHERE UserId = ? AND ConnectorKey = ? LIMIT 1 FOR UPDATE`, userID, connectorKey))
		if errors.Is(err, sql.ErrNoRows) {
			return ErrPrincipalNotFound
		}
		if err != nil {
			return err
		}
		if outcome.principal.Generation != expectedGeneration {
			return ErrGenerationMismatch
		}
		outcome.token, err = scanToken(tx.QueryRowContext(ctx, tokenSelect+`
 WHERE ConnectedPrincipalId = ? LIMIT 1 FOR UPDATE`, outcome.principal.ID))
		if err != nil && !errors.Is(err, sql.ErrNoRows) {
			return err
		}
		disconnectedAt := any(nil)
		if state == PrincipalDisconnected {
			disconnectedAt = now
		}
		if outcome.principal.State == PrincipalConnected {
			result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectedPrincipal
   SET Generation = Generation + 1, State = ?, DisconnectedAt = ?, LastModifiedDate = ?
 WHERE Id = ? AND UserId = ? AND ConnectorKey = ? AND Generation = ?`,
				state, disconnectedAt, now, outcome.principal.ID, userID, connectorKey, outcome.principal.Generation)
			if err != nil {
				return err
			}
			won, err := exactlyOneRow(result)
			if err != nil {
				return err
			}
			if !won {
				return ErrGenerationMismatch
			}
		} else if outcome.principal.State == PrincipalReconnectRequired && state == PrincipalDisconnected {
			result, updateErr := tx.ExecContext(ctx, `
UPDATE _System_ConnectedPrincipal
   SET Generation = Generation + 1, State = ?, DisconnectedAt = ?, LastModifiedDate = ?
 WHERE Id = ? AND UserId = ? AND ConnectorKey = ? AND Generation = ? AND State = ?`,
				PrincipalDisconnected, now, now, outcome.principal.ID, userID, connectorKey,
				outcome.principal.Generation, PrincipalReconnectRequired)
			if updateErr != nil {
				return updateErr
			}
			won, updateErr := exactlyOneRow(result)
			if updateErr != nil {
				return updateErr
			}
			if !won {
				return ErrGenerationMismatch
			}
		} else if outcome.principal.State != PrincipalDisconnected && outcome.principal.State != PrincipalReconnectRequired {
			return ErrEffectStateInvalid
		}
		if outcome.token == nil {
			return nil
		}
		if err := validateTokenEffectShape(*outcome.token); err != nil {
			return err
		}
		switch outcome.token.OAuthEffectState {
		case EffectRefreshDispatched:
			if outcome.token.RefreshDispatchStartedAt == nil || outcome.token.RefreshOwnerID == "" ||
				outcome.token.RefreshLeaseExpiresAt == nil || outcome.token.PendingTokenCiphertext != "" {
				return ErrEffectStateInvalid
			}
			return ErrRefreshInProgress
		case EffectRefreshResultStaged:
			result, err := tx.ExecContext(ctx, `UPDATE _System_ConnectedPrincipalToken
	SET OAuthEffectState = ?, OAuthEffectVersion = OAuthEffectVersion + 1, LastModifiedDate = ?
 WHERE ConnectedPrincipalId = ? AND TokenVersion = ? AND OAuthEffectVersion = ?
	   AND OAuthEffectState = ? AND RefreshOwnerId IS NULL AND RefreshLeaseExpiresAt IS NULL
	   AND RefreshDispatchStartedAt IS NOT NULL AND PendingTokenCiphertext = ?`,
				EffectManualRequired, now, outcome.principal.ID, outcome.token.TokenVersion,
				outcome.token.OAuthEffectVersion, EffectRefreshResultStaged, outcome.token.PendingTokenCiphertext)
			if err != nil {
				return err
			}
			won, err := exactlyOneRow(result)
			if err != nil {
				return err
			}
			if !won {
				return ErrOwnershipLost
			}
			outcome.token.OAuthEffectState = EffectManualRequired
			outcome.token.OAuthEffectVersion++
			return nil
		case EffectNone:
			result, err := tx.ExecContext(ctx, `UPDATE _System_ConnectedPrincipalToken
   SET OAuthEffectState = ?, OAuthEffectVersion = OAuthEffectVersion + 1,
       PendingTokenCiphertext = NULL, RefreshOwnerId = NULL, RefreshLeaseExpiresAt = NULL,
       RefreshDispatchStartedAt = NULL, LastModifiedDate = ?
	WHERE ConnectedPrincipalId = ? AND TokenVersion = ? AND OAuthEffectVersion = ?
	   AND OAuthEffectState = ? AND RefreshOwnerId IS NULL AND RefreshLeaseExpiresAt IS NULL
	   AND RefreshDispatchStartedAt IS NULL AND PendingTokenCiphertext IS NULL`, EffectRevokePending, now,
				outcome.principal.ID, outcome.token.TokenVersion, outcome.token.OAuthEffectVersion, EffectNone)
			if err != nil {
				return err
			}
			won, err := exactlyOneRow(result)
			if err != nil {
				return err
			}
			if !won {
				return ErrOwnershipLost
			}
			return nil
		case EffectRefreshClaimed:
			result, err := tx.ExecContext(ctx, `UPDATE _System_ConnectedPrincipalToken
   SET OAuthEffectState = ?, OAuthEffectVersion = OAuthEffectVersion + 1,
       PendingTokenCiphertext = NULL, RefreshOwnerId = NULL, RefreshLeaseExpiresAt = NULL,
       RefreshDispatchStartedAt = NULL, LastModifiedDate = ?
	WHERE ConnectedPrincipalId = ? AND TokenVersion = ? AND OAuthEffectVersion = ?
	   AND OAuthEffectState = ? AND RefreshOwnershipEpoch = ?
	   AND BINARY RefreshOwnerId = BINARY ? AND RefreshLeaseExpiresAt = ?
	   AND RefreshDispatchStartedAt IS NULL AND PendingTokenCiphertext IS NULL`, EffectRevokePending, now,
				outcome.principal.ID, outcome.token.TokenVersion, outcome.token.OAuthEffectVersion,
				EffectRefreshClaimed, outcome.token.RefreshOwnershipEpoch, outcome.token.RefreshOwnerID,
				outcome.token.RefreshLeaseExpiresAt)
			if err != nil {
				return err
			}
			won, err := exactlyOneRow(result)
			if err != nil {
				return err
			}
			if !won {
				return ErrOwnershipLost
			}
			return nil
		case EffectRevokePending, EffectRevokeClaimed, EffectRevokeDispatched,
			EffectRevokeSucceeded, EffectManualRequired:
			return nil
		default:
			return ErrEffectStateInvalid
		}
	}()
	if transactionErr != nil {
		_ = tx.Rollback()
		return disconnectResult{}, joinDisconnectUnconfirmedErrors(transactionErr, cleanup())
	}
	if err := tx.Commit(); err != nil {
		_ = tx.Rollback()
		return disconnectResult{}, joinDisconnectUnconfirmedErrors(err, cleanup())
	}
	outcome.localCommitConfirmed = true
	if err := cleanup(); err != nil {
		outcome.postCommitCleanupFailed = errDisconnectPostCommitCleanup
	}
	return outcome, nil
}

func joinDisconnectUnconfirmedErrors(primary, cleanup error) error {
	return joinOAuthCleanupErrors(primary, cleanup, errDisconnectUnconfirmedCleanup)
}

func joinOAuthCleanupErrors(primary, cleanup, sanitizedCleanup error) error {
	if cleanup == nil {
		return primary
	}
	return errors.Join(primary, sanitizedCleanup)
}

func lockUser(ctx context.Context, tx *sql.Tx, userID string) error {
	var id string
	err := tx.QueryRowContext(ctx, `SELECT Id FROM User WHERE Id = ? LIMIT 1 FOR UPDATE`, userID).Scan(&id)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrAuthorizationDenied
	}
	return err
}

func withPessimisticTx(ctx context.Context, db *sql.DB, fn func(*sql.Tx) error) (err error) {
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, db)
	if err != nil {
		return err
	}
	defer func() {
		_ = tx.Rollback()
		err = joinOAuthCleanupErrors(err, cleanup(), errOAuthRepositoryTransactionCleanup)
	}()
	if err := fn(tx); err != nil {
		return err
	}
	return tx.Commit()
}

const authorizationSelect = `
SELECT Id, StateDigest, UserId, ConnectorKey, ExpectedGeneration, RedirectUri,
	       CodeVerifierCiphertext, RequestedScopes, ExpiresAt, ConsumedAt, OAuthEffectState,
	       OAuthEffectVersion, OAuthActionOwnerId, OAuthActionEpoch, OAuthActionLeaseExpiresAt,
	       OAuthExchangeDispatchedAt, PendingTokenCiphertext, CreatedDate, LastModifiedDate
  FROM _System_ConnectedAuthorizationRequest`

const principalSelect = `
SELECT Id, UserId, ConnectorKey, Generation, Issuer, Subject, PrincipalAddress, DisplayName,
       State, ConnectedAt, DisconnectedAt, CreatedDate, LastModifiedDate
  FROM _System_ConnectedPrincipal`

const tokenSelect = `
SELECT ConnectedPrincipalId, PrincipalGeneration, AccessTokenCiphertext, RefreshTokenCiphertext,
	       TokenType, GrantedScopes, ExpiresAt, EncryptionKeyId, TokenVersion, RefreshOwnershipEpoch,
	       RefreshOwnerId, RefreshLeaseExpiresAt, RefreshDispatchStartedAt, OAuthEffectState,
	       OAuthEffectVersion, PendingTokenCiphertext, CreatedDate, LastModifiedDate
  FROM _System_ConnectedPrincipalToken`

type rowScanner interface {
	Scan(...any) error
}

func scanAuthorization(row rowScanner) (*authorizationRecord, error) {
	var record authorizationRecord
	var owner, pending sql.NullString
	var consumed, lease, dispatched sql.NullTime
	if err := row.Scan(&record.ID, &record.StateDigest, &record.UserID, &record.ConnectorKey,
		&record.ExpectedGeneration, &record.RedirectURI, &record.CodeVerifierCiphertext,
		&record.RequestedScopes, &record.ExpiresAt, &consumed, &record.OAuthEffectState,
		&record.OAuthEffectVersion, &owner, &record.OAuthActionEpoch, &lease, &dispatched,
		&pending, &record.CreatedDate,
		&record.LastModifiedDate); err != nil {
		return nil, err
	}
	if consumed.Valid {
		record.ConsumedAt = &consumed.Time
	}
	record.OAuthActionOwnerID = owner.String
	record.PendingTokenCiphertext = pending.String
	if lease.Valid {
		record.OAuthActionLeaseExpiresAt = &lease.Time
	}
	if dispatched.Valid {
		record.OAuthExchangeDispatchedAt = &dispatched.Time
	}
	return &record, nil
}

func scanPrincipal(row rowScanner) (*ConnectedPrincipal, error) {
	var principal ConnectedPrincipal
	var displayName sql.NullString
	var disconnected sql.NullTime
	if err := row.Scan(&principal.ID, &principal.UserID, &principal.ConnectorKey, &principal.Generation,
		&principal.Issuer, &principal.Subject, &principal.PrincipalAddress, &displayName,
		&principal.State, &principal.ConnectedAt, &disconnected, &principal.CreatedDate,
		&principal.LastModifiedDate); err != nil {
		return nil, err
	}
	principal.DisplayName = displayName.String
	if disconnected.Valid {
		principal.DisconnectedAt = &disconnected.Time
	}
	return &principal, nil
}

func scanToken(row rowScanner) (*tokenRecord, error) {
	var token tokenRecord
	var refresh, owner, pending sql.NullString
	var expires, lease, dispatched sql.NullTime
	if err := row.Scan(&token.ConnectedPrincipalID, &token.PrincipalGeneration,
		&token.AccessTokenCiphertext, &refresh, &token.TokenType, &token.GrantedScopes,
		&expires, &token.EncryptionKeyID, &token.TokenVersion, &token.RefreshOwnershipEpoch,
		&owner, &lease, &dispatched, &token.OAuthEffectState, &token.OAuthEffectVersion,
		&pending, &token.CreatedDate, &token.LastModifiedDate); err != nil {
		return nil, err
	}
	token.RefreshTokenCiphertext = refresh.String
	token.RefreshOwnerID = owner.String
	token.PendingTokenCiphertext = pending.String
	if expires.Valid {
		token.ExpiresAt = &expires.Time
	}
	if lease.Valid {
		token.RefreshLeaseExpiresAt = &lease.Time
	}
	if dispatched.Valid {
		token.RefreshDispatchStartedAt = &dispatched.Time
	}
	return &token, nil
}

func nullableString(value string) any {
	if strings.TrimSpace(value) == "" {
		return nil
	}
	return value
}

func nullableTime(value *time.Time) any {
	if value == nil {
		return nil
	}
	return *value
}

func exactlyOneRow(result sql.Result) (bool, error) {
	rows, err := result.RowsAffected()
	if err != nil {
		return false, err
	}
	return rows == 1, nil
}

func classifyPrincipalWriteError(err error) error {
	if err == nil {
		return nil
	}
	message := strings.ToLower(err.Error())
	if strings.Contains(message, "duplicate") || strings.Contains(message, "1062") {
		return fmt.Errorf("%w", ErrPrincipalOwnership)
	}
	return err
}