upstream-ory-kratos/persistence/sql/persister_pending_traits_change.go

// Copyright © 2026 Ory Corp
// SPDX-License-Identifier: Apache-2.0

package sql

import (
	"context"
	"time"

	"github.com/gofrs/uuid"
	"github.com/pkg/errors"

	"github.com/ory/kratos/identity"
	"github.com/ory/kratos/persistence/sql/update"
	"github.com/ory/x/otelx"
	"github.com/ory/x/sqlcon"
)

var _ identity.PendingTraitsChangePersister = new(Persister)

func (p *Persister) CreatePendingTraitsChange(ctx context.Context, ptc *identity.PendingTraitsChange) (err error) {
	ctx, span := p.r.Tracer(ctx).Tracer().Start(ctx, "persistence.sql.CreatePendingTraitsChange")
	defer otelx.End(span, &err)

	ptc.ID = uuid.Must(uuid.NewV4())
	ptc.NID = p.NetworkID(ctx)
	ptc.CreatedAt = time.Now().UTC().Truncate(time.Microsecond)
	ptc.UpdatedAt = ptc.CreatedAt

	if err := sqlcon.HandleError(p.GetConnection(ctx).Create(ptc)); err != nil {
		if !errors.Is(err, sqlcon.ErrUniqueViolation()) {
			return err
		}

		// Race: another concurrent request created a pending change after our delete.
		// Delete again and retry once.
		if err := p.DeletePendingTraitsChangesByIdentity(ctx, ptc.IdentityID); err != nil {
			return err
		}
		// TODO(jonas): We could do this on the SQL level with an ON CONFLICT DO UPDATE, but that would require a more complex query and we want to keep the logic in Go for better observability and error handling.
		if err := sqlcon.HandleError(p.GetConnection(ctx).Create(ptc)); err != nil {
			return err
		}
	}
	return nil
}

func (p *Persister) GetPendingTraitsChangeByVerificationFlow(ctx context.Context, flowID uuid.UUID) (_ *identity.PendingTraitsChange, err error) {
	ctx, span := p.r.Tracer(ctx).Tracer().Start(ctx, "persistence.sql.GetPendingTraitsChangeByVerificationFlow")
	defer otelx.End(span, &err)

	var ptc identity.PendingTraitsChange
	if err := p.GetConnection(ctx).
		Where("nid = ? AND verification_flow_id = ? AND status = ?",
			p.NetworkID(ctx), flowID, identity.PendingTraitsChangeStatusPending).
		First(&ptc); err != nil {
		return nil, sqlcon.HandleError(err)
	}
	return &ptc, nil
}

func (p *Persister) DeletePendingTraitsChangesByIdentity(ctx context.Context, identityID uuid.UUID) (err error) {
	ctx, span := p.r.Tracer(ctx).Tracer().Start(ctx, "persistence.sql.DeletePendingTraitsChangesByIdentity")
	defer otelx.End(span, &err)

	return sqlcon.HandleError(
		p.GetConnection(ctx).RawQuery(
			"DELETE FROM identity_pending_traits_changes WHERE nid = ? AND identity_id = ? AND status = 'pending'",
			p.NetworkID(ctx),
			identityID,
		).Exec(),
	)
}

func (p *Persister) UpdatePendingTraitsChange(ctx context.Context, ptc *identity.PendingTraitsChange) (err error) {
	ctx, span := p.r.Tracer(ctx).Tracer().Start(ctx, "persistence.sql.UpdatePendingTraitsChange")
	defer otelx.End(span, &err)

	ptc.UpdatedAt = time.Now().UTC().Truncate(time.Microsecond)
	return update.Generic(ctx, p.GetConnection(ctx), p.r.Tracer(ctx).Tracer(), ptc)
}