IAM: Add email, login field validation to User create/update API (#112391)

* wip

* wip

* wip

(cherry picked from commit 8cedf25892)

* Search seems to be working, the validation is still wip

* Use keyword.Name analyzer for Filterable fields

* Only string fields should be indexed with keyword analyzer

* Change search query for email and login fields to use term query
* Remove unnecessary Exact from the resource protobuf definitions

Co-Authored-By: Ryan McKinley <ryantxu@gmail.com>

* Add legacy search support to the API

* Tests for legacy search, validate and integration tests for user

* Lint

* Add snapshot tests to userDocumentBuilder

* Address CodeQL issues

* Improvements, handle Mode2, tests should pass

* Change default limit from 0 to 1 for requests

* Cleanup

* Add fixme

* Update pkg/registry/apis/iam/register.go

Co-authored-by: Stephanie Hingtgen <stephanie.hingtgen@grafana.com>

* Update pkg/registry/apis/iam/user/legacy_search.go

Co-authored-by: Stephanie Hingtgen <stephanie.hingtgen@grafana.com>

---------

Co-authored-by: Ryan McKinley <ryantxu@gmail.com>
Co-authored-by: Stephanie Hingtgen <stephanie.hingtgen@grafana.com>

pkg/registry/apis/iam/models.go

@@ -13,7 +13,9 @@ import (
 	"github.com/grafana/grafana/pkg/services/authz/zanzana"
 	"github.com/grafana/grafana/pkg/services/featuremgmt"
 	"github.com/grafana/grafana/pkg/services/ssosettings"
+	"github.com/grafana/grafana/pkg/storage/legacysql/dualwrite"
 	"github.com/grafana/grafana/pkg/storage/unified/resource"
+	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
 )
 
 var _ builder.APIGroupBuilder = (*IdentityAccessManagementAPIBuilder)(nil)
@@ -59,6 +61,10 @@ type IdentityAccessManagementAPIBuilder struct {
 	reg    prometheus.Registerer
 	logger log.Logger
 
+	dual             dualwrite.Service
+	unified          resource.ResourceClient
+	userSearchClient resourcepb.ResourceIndexClient
+
 	// non-k8s api route
 	display *user.LegacyDisplayREST
 

pkg/registry/apis/iam/register.go

@@ -42,7 +42,9 @@ import (
 	"github.com/grafana/grafana/pkg/services/authz/zanzana"
 	"github.com/grafana/grafana/pkg/services/featuremgmt"
 	"github.com/grafana/grafana/pkg/services/ssosettings"
+	legacyuser "github.com/grafana/grafana/pkg/services/user"
 	"github.com/grafana/grafana/pkg/storage/legacysql"
+	"github.com/grafana/grafana/pkg/storage/legacysql/dualwrite"
 	"github.com/grafana/grafana/pkg/storage/unified/apistore"
 	"github.com/grafana/grafana/pkg/storage/unified/resource"
 )
@@ -61,6 +63,9 @@ func RegisterAPIService(
 	coreRolesStorage CoreRoleStorageBackend,
 	rolesStorage RoleStorageBackend,
 	roleBindingsStorage RoleBindingStorageBackend,
+	dual dualwrite.Service,
+	unified resource.ResourceClient,
+	userService legacyuser.Service,
 ) (*IdentityAccessManagementAPIBuilder, error) {
 	dbProvider := legacysql.NewDatabaseProvider(sql)
 	store := legacy.NewLegacySQLStores(dbProvider)
@@ -85,6 +90,9 @@ func RegisterAPIService(
 		logger:                     log.New("iam.apis"),
 		features:                   features,
 		enableDualWriter:           true,
+		dual:                       dual,
+		unified:                    unified,
+		userSearchClient:           resource.NewSearchClient(dualwrite.NewSearchAdapter(dual), iamv0.UserResourceInfo.GroupResource(), unified, user.NewUserLegacySearchClient(userService), features),
 	}
 	apiregistration.RegisterAPI(builder)
 
@@ -380,7 +388,7 @@ func (b *IdentityAccessManagementAPIBuilder) Validate(ctx context.Context, a adm
 	case admission.Create:
 		switch typedObj := a.GetObject().(type) {
 		case *iamv0.User:
-			return user.ValidateOnCreate(ctx, typedObj)
+			return user.ValidateOnCreate(ctx, b.userSearchClient, typedObj)
 		case *iamv0.ServiceAccount:
 			return serviceaccount.ValidateOnCreate(ctx, typedObj)
 		case *iamv0.Team:
@@ -398,7 +406,7 @@ func (b *IdentityAccessManagementAPIBuilder) Validate(ctx context.Context, a adm
 			if !ok {
 				return fmt.Errorf("expected old object to be a User, got %T", oldUserObj)
 			}
-			return user.ValidateOnUpdate(ctx, oldUserObj, typedObj)
+			return user.ValidateOnUpdate(ctx, b.userSearchClient, oldUserObj, typedObj)
 		case *iamv0.ResourcePermission:
 			return resourcepermission.ValidateCreateAndUpdateInput(ctx, typedObj)
 		case *iamv0.Team:

pkg/registry/apis/iam/user/legacy_search.go

@@ -0,0 +1,166 @@
+package user
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"math"
+
+	"google.golang.org/grpc"
+
+	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/services/user"
+	res "github.com/grafana/grafana/pkg/storage/unified/resource"
+	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
+	"github.com/grafana/grafana/pkg/storage/unified/search"
+)
+
+const (
+	UserResource      = "users"
+	UserResourceGroup = "iam.grafana.com"
+)
+
+// UserLegacySearchClient is a client for searching for users in the legacy search engine.
+type UserLegacySearchClient struct {
+	resourcepb.ResourceIndexClient
+	userService user.Service
+	log         *slog.Logger
+}
+
+// NewUserLegacySearchClient creates a new UserLegacySearchClient.
+func NewUserLegacySearchClient(userService user.Service) *UserLegacySearchClient {
+	return &UserLegacySearchClient{
+		userService: userService,
+		log:         slog.Default().With("logger", "legacy-user-search-client"),
+	}
+}
+
+// Search searches for users in the legacy search engine.
+// It only supports exact matching for title, login, or email.
+// FIXME: This implementation only supports a single field query and will be extended in the future.
+func (c *UserLegacySearchClient) Search(ctx context.Context, req *resourcepb.ResourceSearchRequest, _ ...grpc.CallOption) (*resourcepb.ResourceSearchResponse, error) {
+	signedInUser, err := identity.GetRequester(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	if req.Limit > 100 {
+		req.Limit = 100
+	}
+	if req.Limit <= 0 {
+		req.Limit = 1
+	}
+
+	if req.Page > math.MaxInt32 || req.Page < 0 {
+		return nil, fmt.Errorf("invalid page number: %d", req.Page)
+	}
+
+	query := &user.SearchUsersQuery{
+		SignedInUser: signedInUser,
+		Limit:        int(req.Limit),
+		Page:         int(req.Page),
+	}
+
+	var title, login, email string
+	for _, field := range req.Options.Fields {
+		vals := field.GetValues()
+		if len(vals) != 1 {
+			c.log.Warn("only single value fields are supported for legacy search, using first value", "field", field.Key, "values", vals)
+		}
+		switch field.Key {
+		case res.SEARCH_FIELD_TITLE:
+			title = vals[0]
+		case "fields.login":
+			login = vals[0]
+		case "fields.email":
+			email = vals[0]
+		}
+	}
+
+	if title == "" && login == "" && email == "" {
+		return nil, fmt.Errorf("at least one of title, login, or email must be provided for the query")
+	}
+
+	// The user store's Search method combines these into an OR.
+	// For legacy search we can only supply one.
+	if title != "" {
+		query.Query = title
+	} else if login != "" {
+		query.Query = login
+	} else {
+		query.Query = email
+	}
+
+	columns := getColumns(req.Fields)
+	list := &resourcepb.ResourceSearchResponse{
+		Results: &resourcepb.ResourceTable{
+			Columns: columns,
+		},
+	}
+
+	res, err := c.userService.Search(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, u := range res.Users {
+		cells := createBaseCells(u, req.Fields)
+		list.Results.Rows = append(list.Results.Rows, &resourcepb.ResourceTableRow{
+			Key:   getResourceKey(u, req.Options.Key.Namespace),
+			Cells: cells,
+		})
+	}
+
+	list.TotalHits = res.TotalCount
+	return list, nil
+}
+
+func getResourceKey(item *user.UserSearchHitDTO, namespace string) *resourcepb.ResourceKey {
+	return &resourcepb.ResourceKey{
+		Namespace: namespace,
+		Group:     UserResourceGroup,
+		Resource:  UserResource,
+		Name:      item.UID,
+	}
+}
+
+func getColumns(fields []string) []*resourcepb.ResourceTableColumnDefinition {
+	columns := defaultColumns()
+	for _, field := range fields {
+		switch field {
+		case "email":
+			columns = append(columns, search.TableColumnDefinitions[search.USER_EMAIL])
+		case "login":
+			columns = append(columns, search.TableColumnDefinitions[search.USER_LOGIN])
+		}
+	}
+	return columns
+}
+
+func createBaseCells(u *user.UserSearchHitDTO, fields []string) [][]byte {
+	cells := createDefaultCells(u)
+	for _, field := range fields {
+		switch field {
+		case "email":
+			cells = append(cells, []byte(u.Email))
+		case "login":
+			cells = append(cells, []byte(u.Login))
+		}
+	}
+	return cells
+}
+
+func createDefaultCells(u *user.UserSearchHitDTO) [][]byte {
+	return [][]byte{
+		[]byte(u.UID),
+		[]byte(u.Name),
+	}
+}
+
+func defaultColumns() []*resourcepb.ResourceTableColumnDefinition {
+	searchFields := res.StandardSearchFields()
+	return []*resourcepb.ResourceTableColumnDefinition{
+		searchFields.Field(res.SEARCH_FIELD_NAME),
+		searchFields.Field(res.SEARCH_FIELD_TITLE),
+	}
+}

pkg/registry/apis/iam/user/legacy_search_fake.go

@@ -0,0 +1,57 @@
+package user
+
+import (
+	"context"
+
+	"google.golang.org/grpc"
+
+	"github.com/grafana/grafana/pkg/services/user"
+	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
+)
+
+// FakeUserLegacySearchClient is a fake implementation of UserLegacySearchClient for testing.
+type FakeUserLegacySearchClient struct {
+	resourcepb.ResourceIndexClient
+	SearchFunc func(ctx context.Context, req *resourcepb.ResourceSearchRequest, opts ...grpc.CallOption) (*resourcepb.ResourceSearchResponse, error)
+	Users      []*user.UserSearchHitDTO
+}
+
+// Search calls the underlying SearchFunc or simulates a search over the Users slice.
+func (c *FakeUserLegacySearchClient) Search(ctx context.Context, req *resourcepb.ResourceSearchRequest, opts ...grpc.CallOption) (*resourcepb.ResourceSearchResponse, error) {
+	if c.SearchFunc != nil {
+		return c.SearchFunc(ctx, req, opts...)
+	}
+
+	// Basic filtering for testing purposes
+	var filteredUsers []*user.UserSearchHitDTO
+	var queryValue string
+
+	for _, field := range req.Options.Fields {
+		if len(field.Values) > 0 {
+			queryValue = field.Values[0]
+			break
+		}
+	}
+
+	for _, u := range c.Users {
+		if u.Login == queryValue || u.Email == queryValue {
+			filteredUsers = append(filteredUsers, u)
+		}
+	}
+
+	rows := make([]*resourcepb.ResourceTableRow, 0, len(filteredUsers))
+	for _, u := range filteredUsers {
+		rows = append(rows, &resourcepb.ResourceTableRow{
+			Key:   getResourceKey(u, req.Options.Key.Namespace),
+			Cells: createBaseCells(u, req.Fields),
+		})
+	}
+
+	return &resourcepb.ResourceSearchResponse{
+		Results: &resourcepb.ResourceTable{
+			Columns: getColumns(req.Fields),
+			Rows:    rows,
+		},
+		TotalHits: int64(len(filteredUsers)),
+	}, nil
+}

pkg/registry/apis/iam/user/legacy_search_test.go

@@ -0,0 +1,148 @@
+package user
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/services/user"
+	"github.com/grafana/grafana/pkg/services/user/usertest"
+	res "github.com/grafana/grafana/pkg/storage/unified/resource"
+	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
+)
+
+func TestUserLegacySearchClient_Search(t *testing.T) {
+	t.Run("should return error if no query fields are provided", func(t *testing.T) {
+		mockUserService := usertest.NewMockService(t)
+		client := NewUserLegacySearchClient(mockUserService)
+		ctx := identity.WithRequester(context.Background(), &user.SignedInUser{OrgID: 1, UserID: 1})
+		req := &resourcepb.ResourceSearchRequest{
+			Options: &resourcepb.ListOptions{
+				Key: &resourcepb.ResourceKey{Namespace: "default"},
+			},
+		}
+
+		_, err := client.Search(ctx, req)
+
+		require.Error(t, err)
+		require.Equal(t, "at least one of title, login, or email must be provided for the query", err.Error())
+	})
+
+	testCases := []struct {
+		name          string
+		fieldKey      string
+		fieldValues   []string
+		expectedQuery string
+	}{
+		{
+			name:          "search by title",
+			fieldKey:      res.SEARCH_FIELD_TITLE,
+			fieldValues:   []string{"test user"},
+			expectedQuery: "test user",
+		},
+		{
+			name:          "search by title (multiple values)",
+			fieldKey:      res.SEARCH_FIELD_TITLE,
+			fieldValues:   []string{"user1", "user2"},
+			expectedQuery: "user1",
+		},
+		{
+			name:          "search by login",
+			fieldKey:      "fields.login",
+			fieldValues:   []string{"testlogin"},
+			expectedQuery: "testlogin",
+		},
+		{
+			name:          "search by email",
+			fieldKey:      "fields.email",
+			fieldValues:   []string{"test@example.com"},
+			expectedQuery: "test@example.com",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			mockUserService := usertest.NewMockService(t)
+			client := NewUserLegacySearchClient(mockUserService)
+			ctx := identity.WithRequester(context.Background(), &user.SignedInUser{OrgID: 1, UserID: 1})
+			req := &resourcepb.ResourceSearchRequest{
+				Limit: 10,
+				Page:  1,
+				Options: &resourcepb.ListOptions{
+					Key: &resourcepb.ResourceKey{Namespace: "default"},
+					Fields: []*resourcepb.Requirement{
+						{Key: tc.fieldKey, Values: tc.fieldValues},
+					},
+				},
+				Fields: []string{"email", "login"},
+			}
+
+			mockUsers := []*user.UserSearchHitDTO{
+				{ID: 1, UID: "uid1", Name: "Test User 1", Email: "test1@example.com", Login: "testlogin1"},
+			}
+
+			mockUserService.On("Search", mock.Anything, mock.MatchedBy(func(q *user.SearchUsersQuery) bool {
+				return q.Query == tc.expectedQuery && q.Limit == 10 && q.Page == 1
+			})).Return(&user.SearchUserQueryResult{
+				Users:      mockUsers,
+				TotalCount: 1,
+			}, nil)
+
+			resp, err := client.Search(ctx, req)
+			require.NoError(t, err)
+			require.NotNil(t, resp)
+			require.Equal(t, int64(1), resp.TotalHits)
+			require.Len(t, resp.Results.Rows, 1)
+
+			// Verify columns
+			expectedColumns := getColumns(req.Fields)
+			require.Equal(t, len(expectedColumns), len(resp.Results.Columns))
+			for i, col := range resp.Results.Columns {
+				require.Equal(t, expectedColumns[i].Name, col.Name)
+			}
+
+			// Verify rows
+			for i, u := range mockUsers {
+				row := resp.Results.Rows[i]
+				require.Equal(t, "default", row.Key.Namespace)
+				require.Equal(t, UserResourceGroup, row.Key.Group)
+				require.Equal(t, UserResource, row.Key.Resource)
+				require.Equal(t, u.UID, row.Key.Name)
+
+				expectedCells := createBaseCells(&user.UserSearchHitDTO{
+					UID:   u.UID,
+					Name:  u.Name,
+					Email: u.Email,
+					Login: u.Login,
+				}, req.Fields)
+				require.Equal(t, expectedCells, row.Cells)
+			}
+		})
+	}
+
+	t.Run("title should have precedence over login and email", func(t *testing.T) {
+		mockUserService := usertest.NewMockService(t)
+		client := NewUserLegacySearchClient(mockUserService)
+		ctx := identity.WithRequester(context.Background(), &user.SignedInUser{OrgID: 1, UserID: 1})
+		req := &resourcepb.ResourceSearchRequest{
+			Options: &resourcepb.ListOptions{
+				Key: &resourcepb.ResourceKey{Namespace: "default"},
+				Fields: []*resourcepb.Requirement{
+					{Key: res.SEARCH_FIELD_TITLE, Values: []string{"title"}},
+					{Key: "fields.login", Values: []string{"login"}},
+					{Key: "fields.email", Values: []string{"email"}},
+				},
+			},
+		}
+
+		mockUserService.On("Search", mock.Anything, mock.MatchedBy(func(q *user.SearchUsersQuery) bool {
+			return q.Query == "title"
+		})).Return(&user.SearchUserQueryResult{Users: []*user.UserSearchHitDTO{}, TotalCount: 0}, nil)
+
+		_, err := client.Search(ctx, req)
+		require.NoError(t, err)
+	})
+}

pkg/registry/apis/iam/user/validate.go

@@ -5,13 +5,15 @@ import (
 	"fmt"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/selection"
 
 	"github.com/grafana/authlib/types"
 	iamv0alpha1 "github.com/grafana/grafana/apps/iam/pkg/apis/iam/v0alpha1"
 	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
 )
 
-func ValidateOnCreate(ctx context.Context, obj *iamv0alpha1.User) error {
+func ValidateOnCreate(ctx context.Context, userSearchClient resourcepb.ResourceIndexClient, obj *iamv0alpha1.User) error {
 	requester, err := identity.GetRequester(ctx)
 	if err != nil {
 		return apierrors.NewUnauthorized("no identity found")
@@ -28,27 +30,22 @@ func ValidateOnCreate(ctx context.Context, obj *iamv0alpha1.User) error {
 		return apierrors.NewBadRequest("user must have either login or email")
 	}
 
-	err = validateRole(obj)
-	if err != nil {
+	if err := validateRole(obj); err != nil {
+		return err
+	}
+
+	if err := validateEmail(ctx, userSearchClient, requester.GetNamespace(), obj.Name, obj.Spec.Email); err != nil {
+		return err
+	}
+
+	if err := validateLogin(ctx, userSearchClient, requester.GetNamespace(), obj.Name, obj.Spec.Login); err != nil {
 		return err
 	}
 
 	return nil
 }
 
-func validateRole(obj *iamv0alpha1.User) error {
-	if obj.Spec.Role == "" {
-		return apierrors.NewBadRequest("role is required")
-	}
-
-	if !identity.RoleType(obj.Spec.Role).IsValid() {
-		return apierrors.NewBadRequest(fmt.Sprintf("invalid role '%s'", obj.Spec.Role))
-	}
-
-	return nil
-}
-
-func ValidateOnUpdate(ctx context.Context, oldObj, newObj *iamv0alpha1.User) error {
+func ValidateOnUpdate(ctx context.Context, userSearchClient resourcepb.ResourceIndexClient, oldObj, newObj *iamv0alpha1.User) error {
 	requester, err := identity.GetRequester(ctx)
 	if err != nil {
 		return apierrors.NewUnauthorized("no identity found")
@@ -93,10 +90,109 @@ func ValidateOnUpdate(ctx context.Context, oldObj, newObj *iamv0alpha1.User) err
 		return apierrors.NewBadRequest("user must have either login or email")
 	}
 
-	err = validateRole(newObj)
-	if err != nil {
+	if err := validateRole(newObj); err != nil {
 		return err
 	}
 
+	if newObj.Spec.Email != oldObj.Spec.Email {
+		if err := validateEmail(ctx, userSearchClient, requester.GetNamespace(), newObj.Name, newObj.Spec.Email); err != nil {
+			return err
+		}
+	}
+
+	if newObj.Spec.Login != oldObj.Spec.Login {
+		if err := validateLogin(ctx, userSearchClient, requester.GetNamespace(), newObj.Name, newObj.Spec.Login); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
+
+func validateRole(obj *iamv0alpha1.User) error {
+	if obj.Spec.Role == "" {
+		return apierrors.NewBadRequest("role is required")
+	}
+
+	if !identity.RoleType(obj.Spec.Role).IsValid() {
+		return apierrors.NewBadRequest(fmt.Sprintf("invalid role '%s'", obj.Spec.Role))
+	}
+
+	return nil
+}
+
+func validateEmail(ctx context.Context, searchClient resourcepb.ResourceIndexClient, namespace, name, email string) error {
+	req := createUserSearchRequest(namespace, []*resourcepb.Requirement{
+		{
+			Key:      "fields.email",
+			Operator: string(selection.Equals),
+			Values:   []string{email},
+		},
+	}, []string{"name", "email", "login"})
+
+	resp, err := searchClient.Search(ctx, req)
+	if err != nil {
+		return err
+	}
+
+	// FIXME(mgyongyosi): Improve the exact match validation
+
+	if resp.TotalHits > 0 {
+		// If the found user is the same as the one being created/updated, it's not a conflict.
+		// This is required for Mode 2 when the resource is written to LegacyStorage and UnifiedStorage.
+		rows := resp.Results.Rows
+		if len(rows) > 0 && rows[0].Key.Name == name {
+			return nil
+		}
+		return apierrors.NewConflict(iamv0alpha1.UserResourceInfo.GroupResource(),
+			name,
+			fmt.Errorf("email '%s' is already taken", email))
+	}
+
+	return nil
+}
+
+func validateLogin(ctx context.Context, searchClient resourcepb.ResourceIndexClient, namespace, name, login string) error {
+	req := createUserSearchRequest(namespace, []*resourcepb.Requirement{
+		{
+			Key:      "fields.login",
+			Operator: string(selection.Equals),
+			Values:   []string{login},
+		},
+	}, []string{"name", "email", "login"})
+	resp, err := searchClient.Search(ctx, req)
+	if err != nil {
+		return err
+	}
+
+	// FIXME(mgyongyosi): Improve the exact match validation
+
+	if resp.TotalHits > 0 {
+		// If the found user is the same as the one being created/updated, it's not a conflict.
+		// This is required for Mode 2 when the resource is written to LegacyStorage and UnifiedStorage.
+		rows := resp.Results.Rows
+		if len(rows) > 0 && rows[0].Key.Name == name {
+			return nil
+		}
+		return apierrors.NewConflict(iamv0alpha1.UserResourceInfo.GroupResource(),
+			name,
+			fmt.Errorf("login '%s' is already taken", login))
+	}
+
+	return nil
+}
+
+func createUserSearchRequest(namespace string, requirements []*resourcepb.Requirement, fields []string) *resourcepb.ResourceSearchRequest {
+	userGvr := iamv0alpha1.UserResourceInfo.GroupResource()
+	return &resourcepb.ResourceSearchRequest{
+		Options: &resourcepb.ListOptions{
+			Key: &resourcepb.ResourceKey{
+				Group:     userGvr.Group,
+				Resource:  userGvr.Resource,
+				Namespace: namespace,
+			},
+			Fields: requirements,
+		},
+		Fields: fields,
+	}
+}

pkg/registry/apis/iam/user/validate_test.go

@@ -9,6 +9,9 @@ import (
 	"github.com/grafana/authlib/types"
 	iamv0alpha1 "github.com/grafana/grafana/apps/iam/pkg/apis/iam/v0alpha1"
 	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/services/user"
+	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 func TestValidateOnCreate(t *testing.T) {
@@ -16,6 +19,7 @@ func TestValidateOnCreate(t *testing.T) {
 		name          string
 		user          *iamv0alpha1.User
 		requester     *identity.StaticRequester
+		searchClient  resourcepb.ResourceIndexClient
 		expectError   bool
 		errorContains string
 	}{
@@ -31,7 +35,8 @@ func TestValidateOnCreate(t *testing.T) {
 				Type:           types.TypeUser,
 				IsGrafanaAdmin: true,
 			},
-			expectError: false,
+			searchClient: &FakeUserLegacySearchClient{},
+			expectError:  false,
 		},
 		{
 			name: "grafana admin creating another grafana admin",
@@ -46,7 +51,8 @@ func TestValidateOnCreate(t *testing.T) {
 				Type:           types.TypeUser,
 				IsGrafanaAdmin: true,
 			},
-			expectError: false,
+			searchClient: &FakeUserLegacySearchClient{},
+			expectError:  false,
 		},
 		{
 			name: "non-admin trying to create a grafana admin",
@@ -61,6 +67,7 @@ func TestValidateOnCreate(t *testing.T) {
 				Type:           types.TypeUser,
 				IsGrafanaAdmin: false,
 			},
+			searchClient:  &FakeUserLegacySearchClient{},
 			expectError:   true,
 			errorContains: "only grafana admins can create grafana admins",
 		},
@@ -75,6 +82,7 @@ func TestValidateOnCreate(t *testing.T) {
 				Type:           types.TypeUser,
 				IsGrafanaAdmin: false,
 			},
+			searchClient:  &FakeUserLegacySearchClient{},
 			expectError:   true,
 			errorContains: "user must have either login or email",
 		},
@@ -90,13 +98,14 @@ func TestValidateOnCreate(t *testing.T) {
 				Type:           types.TypeUser,
 				IsGrafanaAdmin: false,
 			},
-			expectError: false,
+			searchClient: &FakeUserLegacySearchClient{},
+			expectError:  false,
 		},
 		{
 			name: "user with only email",
 			user: &iamv0alpha1.User{
 				Spec: iamv0alpha1.UserSpec{
-					Email: "test@test.com",
+					Email: "test@example",
 					Role:  "Viewer",
 				},
 			},
@@ -104,7 +113,8 @@ func TestValidateOnCreate(t *testing.T) {
 				Type:           types.TypeUser,
 				IsGrafanaAdmin: false,
 			},
-			expectError: false,
+			searchClient: &FakeUserLegacySearchClient{},
+			expectError:  false,
 		},
 		{
 			name: "user with empty role",
@@ -117,6 +127,7 @@ func TestValidateOnCreate(t *testing.T) {
 				Type:           types.TypeUser,
 				IsGrafanaAdmin: false,
 			},
+			searchClient:  &FakeUserLegacySearchClient{},
 			expectError:   true,
 			errorContains: "role is required",
 		},
@@ -132,6 +143,7 @@ func TestValidateOnCreate(t *testing.T) {
 				Type:           types.TypeUser,
 				IsGrafanaAdmin: false,
 			},
+			searchClient:  &FakeUserLegacySearchClient{},
 			expectError:   true,
 			errorContains: "invalid role 'InvalidRole'",
 		},
@@ -147,7 +159,55 @@ func TestValidateOnCreate(t *testing.T) {
 				Type:           types.TypeUser,
 				IsGrafanaAdmin: true,
 			},
-			expectError: false,
+			searchClient: &FakeUserLegacySearchClient{},
+			expectError:  false,
+		},
+		{
+			name: "user with existing email",
+			user: &iamv0alpha1.User{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "userx",
+				},
+				Spec: iamv0alpha1.UserSpec{
+					Email: "existing@example",
+					Role:  "Viewer",
+				},
+			},
+			requester: &identity.StaticRequester{
+				Type:           types.TypeUser,
+				IsGrafanaAdmin: false,
+			},
+			searchClient: &FakeUserLegacySearchClient{
+				Users: []*user.UserSearchHitDTO{
+					{Email: "existing@example"},
+				},
+			},
+			expectError:   true,
+			errorContains: "email 'existing@example' is already taken",
+		},
+		{
+			name: "user with existing login",
+			user: &iamv0alpha1.User{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "userx",
+				},
+				Spec: iamv0alpha1.UserSpec{
+					Login: "existinguser",
+					Email: "existinguser@example",
+					Role:  "Viewer",
+				},
+			},
+			requester: &identity.StaticRequester{
+				Type:           types.TypeUser,
+				IsGrafanaAdmin: false,
+			},
+			searchClient: &FakeUserLegacySearchClient{
+				Users: []*user.UserSearchHitDTO{
+					{Login: "existinguser"},
+				},
+			},
+			expectError:   true,
+			errorContains: "login 'existinguser' is already taken",
 		},
 	}
 
@@ -158,7 +218,7 @@ func TestValidateOnCreate(t *testing.T) {
 				tt.requester,
 			)
 
-			err := ValidateOnCreate(ctx, tt.user)
+			err := ValidateOnCreate(ctx, tt.searchClient, tt.user)
 
 			if tt.expectError {
 				require.Error(t, err)
@@ -178,6 +238,7 @@ func TestValidateOnUpdate(t *testing.T) {
 		oldUser       *iamv0alpha1.User
 		newUser       *iamv0alpha1.User
 		requester     *identity.StaticRequester
+		searchClient  resourcepb.ResourceIndexClient
 		expectError   bool
 		errorContains string
 	}{
@@ -254,7 +315,7 @@ func TestValidateOnUpdate(t *testing.T) {
 		{
 			name: "update with only login",
 			oldUser: &iamv0alpha1.User{
-				Spec: iamv0alpha1.UserSpec{Email: "test@test.com", Role: "Viewer"},
+				Spec: iamv0alpha1.UserSpec{Email: "test@example", Role: "Viewer"},
 			},
 			newUser: &iamv0alpha1.User{
 				Spec: iamv0alpha1.UserSpec{Login: "testuser", Email: "", Role: "Viewer"},
@@ -263,7 +324,8 @@ func TestValidateOnUpdate(t *testing.T) {
 				Type:           types.TypeUser,
 				IsGrafanaAdmin: true,
 			},
-			expectError: false,
+			searchClient: &FakeUserLegacySearchClient{},
+			expectError:  false,
 		},
 		{
 			name: "update with only email",
@@ -271,13 +333,14 @@ func TestValidateOnUpdate(t *testing.T) {
 				Spec: iamv0alpha1.UserSpec{Login: "testuser", Role: "Viewer"},
 			},
 			newUser: &iamv0alpha1.User{
-				Spec: iamv0alpha1.UserSpec{Login: "", Email: "test@test.com", Role: "Viewer"},
+				Spec: iamv0alpha1.UserSpec{Login: "", Email: "test@example", Role: "Viewer"},
 			},
 			requester: &identity.StaticRequester{
 				Type:           types.TypeUser,
 				IsGrafanaAdmin: true,
 			},
-			expectError: false,
+			searchClient: &FakeUserLegacySearchClient{},
+			expectError:  false,
 		},
 		{
 			name: "service user verifies email",
@@ -408,6 +471,77 @@ func TestValidateOnUpdate(t *testing.T) {
 			},
 			expectError: false,
 		},
+		{
+			name: "update with existing email",
+			oldUser: &iamv0alpha1.User{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "userx",
+				},
+				Spec: iamv0alpha1.UserSpec{Email: "one@example", Role: "Viewer"},
+			},
+			newUser: &iamv0alpha1.User{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "userx",
+				},
+				Spec: iamv0alpha1.UserSpec{Email: "two@example", Role: "Viewer"},
+			},
+			requester: &identity.StaticRequester{
+				Type:           types.TypeUser,
+				IsGrafanaAdmin: true,
+			},
+			searchClient: &FakeUserLegacySearchClient{
+				Users: []*user.UserSearchHitDTO{
+					{Email: "two@example"},
+				},
+			},
+			expectError:   true,
+			errorContains: "email 'two@example' is already taken",
+		},
+		{
+			name: "update with existing login",
+			oldUser: &iamv0alpha1.User{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "userx",
+				},
+				Spec: iamv0alpha1.UserSpec{Login: "one", Role: "Viewer"},
+			},
+			newUser: &iamv0alpha1.User{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "userx",
+				},
+				Spec: iamv0alpha1.UserSpec{Login: "two", Role: "Viewer"},
+			},
+			requester: &identity.StaticRequester{
+				Type:           types.TypeUser,
+				IsGrafanaAdmin: true,
+			},
+			searchClient: &FakeUserLegacySearchClient{
+				Users: []*user.UserSearchHitDTO{
+					{Name: "other", UID: "uid456", Login: "two"},
+				},
+			},
+			expectError:   true,
+			errorContains: "login 'two' is already taken",
+		},
+		{
+			name: "update with no change to login or email",
+			oldUser: &iamv0alpha1.User{
+				Spec: iamv0alpha1.UserSpec{Login: "testuser", Email: "test@example", Role: "Viewer"},
+			},
+			newUser: &iamv0alpha1.User{
+				Spec: iamv0alpha1.UserSpec{Login: "testuser", Email: "test@example", Role: "Editor"},
+			},
+			requester: &identity.StaticRequester{
+				Type:           types.TypeUser,
+				IsGrafanaAdmin: true,
+			},
+			searchClient: &FakeUserLegacySearchClient{
+				Users: []*user.UserSearchHitDTO{
+					{Login: "testuser", Email: "test@example"},
+				},
+			},
+			expectError: false,
+		},
 	}
 
 	for _, tt := range tests {
@@ -417,7 +551,7 @@ func TestValidateOnUpdate(t *testing.T) {
 				tt.requester,
 			)
 
-			err := ValidateOnUpdate(ctx, tt.oldUser, tt.newUser)
+			err := ValidateOnUpdate(ctx, tt.searchClient, tt.oldUser, tt.newUser)
 
 			if tt.expectError {
 				require.Error(t, err)

pkg/server/wire_gen.go

@@ -823,7 +823,7 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api
 	}
 	folderAPIBuilder := folders.RegisterAPIService(cfg, featureToggles, apiserverService, folderimplService, folderPermissionsService, accessControl, acimplService, accessClient, registerer, resourceClient, zanzanaClient)
 	storageBackendImpl := noopstorage.ProvideStorageBackend()
-	identityAccessManagementAPIBuilder, err := iam.RegisterAPIService(featureToggles, apiserverService, ssosettingsimplService, sqlStore, accessControl, accessClient, zanzanaClient, registerer, storageBackendImpl, storageBackendImpl, storageBackendImpl)
+	identityAccessManagementAPIBuilder, err := iam.RegisterAPIService(featureToggles, apiserverService, ssosettingsimplService, sqlStore, accessControl, accessClient, zanzanaClient, registerer, storageBackendImpl, storageBackendImpl, storageBackendImpl, dualwriteService, resourceClient, userService)
 	if err != nil {
 		return nil, err
 	}
@@ -1435,7 +1435,7 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac
 	}
 	folderAPIBuilder := folders.RegisterAPIService(cfg, featureToggles, apiserverService, folderimplService, folderPermissionsService, accessControl, acimplService, accessClient, registerer, resourceClient, zanzanaClient)
 	storageBackendImpl := noopstorage.ProvideStorageBackend()
-	identityAccessManagementAPIBuilder, err := iam.RegisterAPIService(featureToggles, apiserverService, ssosettingsimplService, sqlStore, accessControl, accessClient, zanzanaClient, registerer, storageBackendImpl, storageBackendImpl, storageBackendImpl)
+	identityAccessManagementAPIBuilder, err := iam.RegisterAPIService(featureToggles, apiserverService, ssosettingsimplService, sqlStore, accessControl, accessClient, zanzanaClient, registerer, storageBackendImpl, storageBackendImpl, storageBackendImpl, dualwriteService, resourceClient, userService)
 	if err != nil {
 		return nil, err
 	}

pkg/storage/unified/search/bleve.go

@@ -1542,6 +1542,14 @@ func requirementQuery(req *resourcepb.Requirement, prefix string) (query.Query,
 			return query.NewMatchAllQuery(), nil
 		}
 
+		// FIXME: special case for login and email to use term query only because those fields are using keyword analyzer
+		// This should be fixed by using the info from the schema
+		if (req.Key == "login" || req.Key == "email") && len(req.Values) == 1 {
+			tq := bleve.NewTermQuery(req.Values[0])
+			tq.SetField(prefix + req.Key)
+			return tq, nil
+		}
+
 		if len(req.Values) == 1 {
 			filter := filterValue(req.Key, req.Values[0])
 			return newQuery(req.Key, filter, prefix), nil

pkg/storage/unified/search/bleve_mappings.go

@@ -7,6 +7,7 @@ import (
 	"github.com/blevesearch/bleve/v2/mapping"
 
 	"github.com/grafana/grafana/pkg/storage/unified/resource"
+	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
 )
 
 func GetBleveMappings(fields resource.SearchableDocumentFields) (mapping.IndexMapping, error) {
@@ -21,7 +22,7 @@ func GetBleveMappings(fields resource.SearchableDocumentFields) (mapping.IndexMa
 	return mapper, nil
 }
 
-func getBleveDocMappings(_ resource.SearchableDocumentFields) *mapping.DocumentMapping {
+func getBleveDocMappings(fields resource.SearchableDocumentFields) *mapping.DocumentMapping {
 	mapper := bleve.NewDocumentStaticMapping()
 
 	nameMapping := &mapping.FieldMapping{
@@ -145,6 +146,23 @@ func getBleveDocMappings(_ resource.SearchableDocumentFields) *mapping.DocumentM
 	mapper.AddSubDocumentMapping(resource.SEARCH_FIELD_LABELS, labelMapper)
 
 	fieldMapper := bleve.NewDocumentMapping()
+	if fields != nil {
+		for _, field := range fields.Fields() {
+			def := fields.Field(field)
+
+			// Filterable should use keyword analyzer for exact matches
+			if def.Properties != nil && def.Properties.Filterable && def.Type == resourcepb.ResourceTableColumnDefinition_STRING {
+				keywordMapping := bleve.NewKeywordFieldMapping()
+				keywordMapping.Store = true
+
+				fieldMapper.AddFieldMappingsAt(def.Name, keywordMapping)
+			}
+			// For all other fields, we do nothing.
+			// Bleve will see them at index time and dynamically map them as
+			// numeric, datetime, boolean, or standard text based on their content.
+		}
+	}
+
 	mapper.AddSubDocumentMapping("fields", fieldMapper)
 
 	return mapper

pkg/storage/unified/search/document.go

@@ -65,6 +65,15 @@ func (s *StandardDocumentBuilders) GetDocumentBuilders() ([]resource.DocumentBui
 		}, nil
 	})
 
+	if err != nil {
+		return nil, err
+	}
+
+	users, err := GetUserBuilder()
+	if err != nil {
+		return nil, err
+	}
+
 	return []resource.DocumentBuilderInfo{
 		// The default builder
 		{
@@ -72,5 +81,7 @@ func (s *StandardDocumentBuilders) GetDocumentBuilders() ([]resource.DocumentBui
 		},
 		// Dashboard builder
 		dashboards,
+		// User builder
+		users,
 	}, err
 }

pkg/storage/unified/search/testdata/doc/user-example-out.json

@@ -0,0 +1,17 @@
+{
+  "key": {
+    "namespace": "default",
+    "group": "iam.grafana.app",
+    "resource": "users",
+    "name": "example"
+  },
+  "name": "example",
+  "rv": 1234,
+  "title": "example",
+  "title_ngram": "example",
+  "title_phrase": "example",
+  "fields": {
+    "email": "example@example.com",
+    "login": "example"
+  }
+}

pkg/storage/unified/search/testdata/doc/user-example.json

@@ -0,0 +1,13 @@
+{
+  "apiVersion": "iam.grafana.app/v0alpha1",
+  "kind": "User",
+  "metadata": {
+    "name": "example",
+    "namespace": "default"
+  },
+  "spec": {
+    "login": "example",
+    "email": "example@example.com",
+    "role": "Viewer"
+  }
+}

pkg/storage/unified/search/testdata/doc/user-user-with-login-and-email-out.json

@@ -0,0 +1,14 @@
+{
+  "key": {
+    "namespace": "default",
+    "group": "iam.grafana.app",
+    "resource": "users",
+    "name": "user-with-login-and-email"
+  },
+  "name": "user-with-login-and-email",
+  "rv": 1234,
+  "fields": {
+    "email": "user.one@test.com",
+    "login": "user.one"
+  }
+}

pkg/storage/unified/search/testdata/doc/user-user-with-login-and-email.json

@@ -0,0 +1,11 @@
+{
+    "metadata": {
+        "name": "user-with-login-and-email",
+        "namespace": "default"
+    },
+    "spec": {
+        "login": "user.one",
+        "email": "user.one@test.com",
+        "role": "Viewer"
+    }
+}

pkg/storage/unified/search/testdata/doc/user-user-with-login-only-out.json

@@ -0,0 +1,13 @@
+{
+  "key": {
+    "namespace": "default",
+    "group": "iam.grafana.app",
+    "resource": "users",
+    "name": "user-with-login-only"
+  },
+  "name": "user-with-login-only",
+  "rv": 1234,
+  "fields": {
+    "login": "user.two"
+  }
+}

pkg/storage/unified/search/testdata/doc/user-user-with-login-only.json

@@ -0,0 +1,10 @@
+{
+    "metadata": {
+        "name": "user-with-login-only",
+        "namespace": "default"
+    },
+    "spec": {
+        "login": "user.two",
+        "role": "Viewer"
+    }
+}

pkg/storage/unified/search/user.go

@@ -0,0 +1,81 @@
+package search
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+
+	iamv0 "github.com/grafana/grafana/apps/iam/pkg/apis/iam/v0alpha1"
+	"github.com/grafana/grafana/pkg/apimachinery/utils"
+	"github.com/grafana/grafana/pkg/storage/unified/resource"
+	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
+)
+
+const (
+	USER_EMAIL = "email"
+	USER_LOGIN = "login"
+)
+
+var TableColumnDefinitions = map[string]*resourcepb.ResourceTableColumnDefinition{
+	USER_EMAIL: {
+		Name:        USER_EMAIL,
+		Type:        resourcepb.ResourceTableColumnDefinition_STRING,
+		Description: "The email address of the user",
+		Properties: &resourcepb.ResourceTableColumnDefinition_Properties{
+			UniqueValues: true,
+			Filterable:   true,
+		},
+	},
+	USER_LOGIN: {
+		Name:        USER_LOGIN,
+		Type:        resourcepb.ResourceTableColumnDefinition_STRING,
+		Description: "The login of the user",
+		Properties: &resourcepb.ResourceTableColumnDefinition_Properties{
+			UniqueValues: true,
+			Filterable:   true,
+		},
+	},
+}
+
+func GetUserBuilder() (resource.DocumentBuilderInfo, error) {
+	values := make([]*resourcepb.ResourceTableColumnDefinition, 0, len(TableColumnDefinitions))
+	for _, v := range TableColumnDefinitions {
+		values = append(values, v)
+	}
+	fields, err := resource.NewSearchableDocumentFields(values)
+	return resource.DocumentBuilderInfo{
+		GroupResource: iamv0.UserResourceInfo.GroupResource(),
+		Fields:        fields,
+		Builder:       new(userDocumentBuilder),
+	}, err
+}
+
+var _ resource.DocumentBuilder = new(userDocumentBuilder)
+
+type userDocumentBuilder struct{}
+
+// BuildDocument implements resource.DocumentBuilder.
+func (u *userDocumentBuilder) BuildDocument(ctx context.Context, key *resourcepb.ResourceKey, rv int64, value []byte) (*resource.IndexableDocument, error) {
+	user := &iamv0.User{}
+	err := json.NewDecoder(bytes.NewReader(value)).Decode(user)
+	if err != nil {
+		return nil, err
+	}
+
+	obj, err := utils.MetaAccessor(user)
+	if err != nil {
+		return nil, err
+	}
+
+	doc := resource.NewIndexableDocument(key, rv, obj)
+
+	doc.Fields = make(map[string]any)
+	if user.Spec.Email != "" {
+		doc.Fields[USER_EMAIL] = user.Spec.Email
+	}
+	if user.Spec.Login != "" {
+		doc.Fields[USER_LOGIN] = user.Spec.Login
+	}
+
+	return doc, nil
+}

pkg/storage/unified/search/user_test.go

@@ -0,0 +1,214 @@
+package search_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	iamv0 "github.com/grafana/grafana/apps/iam/pkg/apis/iam/v0alpha1"
+	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/infra/tracing"
+	"github.com/grafana/grafana/pkg/services/user"
+	"github.com/grafana/grafana/pkg/storage/unified/resource"
+	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
+	"github.com/grafana/grafana/pkg/storage/unified/search"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/selection"
+)
+
+func TestUserDocumentBuilder(t *testing.T) {
+	info, err := search.GetUserBuilder()
+	require.NoError(t, err)
+	doSnapshotTests(t, info.Builder, "user", &resourcepb.ResourceKey{
+		Namespace: "default",
+		Group:     "iam.grafana.app",
+		Resource:  "users",
+	}, []string{
+		"user-with-login-and-email",
+		"user-with-login-only",
+	})
+}
+
+func TestUserSearch(t *testing.T) {
+	key := resource.NamespacedResource{
+		Namespace: "default",
+		Group:     iamv0.UserResourceInfo.GroupResource().Group,
+		Resource:  iamv0.UserResourceInfo.GroupResource().Resource,
+	}
+
+	index := newTestUsersIndex(t, 100, 2, func(index resource.ResourceIndex) (int64, error) {
+		return 0, nil
+	})
+	users := []iamv0.User{
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "user1",
+				Namespace: "default",
+			},
+			Spec: iamv0.UserSpec{
+				Login: "user.one",
+				Email: "user.one@test.com",
+				Role:  "Viewer",
+			},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "user2",
+				Namespace: "default",
+			},
+			Spec: iamv0.UserSpec{
+				Login: "user.two",
+				Email: "user.two@test.com",
+				Role:  "Viewer",
+			},
+		},
+	}
+	indexUserDocuments(t, index, key, users)
+
+	// Sanity check - title search
+	checkUserSearchQuery(t, index, newTestsUserQueryWithTitle(key, "user2"), []string{"user2"})
+
+	t.Run("can search users by login", func(t *testing.T) {
+		// Search by login
+		checkUserSearchQuery(t, index, newTestUserQueryWithReqs(key, []*resourcepb.Requirement{
+			{
+				Key:      "fields.login",
+				Operator: string(selection.Equals),
+				Values:   []string{"user.one"},
+			},
+		}), []string{"user1"})
+		checkUserSearchQuery(t, index, newTestUserQueryWithReqs(key, []*resourcepb.Requirement{
+			{
+				Key:      "fields.login",
+				Operator: string(selection.Equals),
+				Values:   []string{"user.two"},
+			},
+		}), []string{"user2"})
+	})
+
+	t.Run("can search users by wildcard login", func(t *testing.T) {
+		checkUserSearchQuery(t, index, newTestUserQueryWithReqs(key, []*resourcepb.Requirement{
+			{
+				Key:      "fields.login",
+				Operator: string(selection.Equals),
+				Values:   []string{"user.*"},
+			},
+		}), []string{"user1", "user2"})
+	})
+
+	t.Run("can search users by email", func(t *testing.T) {
+		// Search by email
+		checkUserSearchQuery(t, index, newTestUserQueryWithReqs(key, []*resourcepb.Requirement{
+			{
+				Key:      "fields.email",
+				Operator: string(selection.Equals),
+				Values:   []string{"user.one@test.com"},
+			},
+		}), []string{"user1"})
+
+		checkUserSearchQuery(t, index, newTestUserQueryWithReqs(key, []*resourcepb.Requirement{
+			{
+				Key:      "fields.email",
+				Operator: string(selection.Equals),
+				Values:   []string{"user.two@test.com"},
+			},
+		}), []string{"user2"})
+	})
+}
+
+func newTestUsersIndex(t testing.TB, threshold int64, size int64, writer resource.BuildFn) resource.ResourceIndex {
+	t.Helper()
+	gr := iamv0.UserResourceInfo.GroupResource()
+	key := &resourcepb.ResourceKey{
+		Namespace: "default",
+		Group:     gr.Group,
+		Resource:  gr.Resource,
+	}
+	backend, err := search.NewBleveBackend(search.BleveOptions{
+		Root:          t.TempDir(),
+		FileThreshold: threshold, // use in-memory for tests
+	}, tracing.NewNoopTracerService(), nil)
+	require.NoError(t, err)
+
+	t.Cleanup(backend.Stop)
+
+	ctx := identity.WithRequester(context.Background(), &user.SignedInUser{Namespace: "ns"})
+
+	info, err := search.GetUserBuilder()
+	require.NoError(t, err)
+
+	index, err := backend.BuildIndex(ctx, resource.NamespacedResource{
+		Namespace: key.Namespace,
+		Group:     key.Group,
+		Resource:  key.Resource,
+	}, size, info.Fields, "test", writer, nil, false)
+	require.NoError(t, err)
+
+	return index
+}
+
+func indexUserDocuments(t *testing.T, index resource.ResourceIndex, key resource.NamespacedResource, users []iamv0.User) {
+	t.Helper()
+	items := make([]*resource.BulkIndexItem, 0, len(users))
+	for _, user := range users {
+		items = append(items, &resource.BulkIndexItem{
+			Action: resource.ActionIndex,
+			Doc: &resource.IndexableDocument{
+				RV:   1,
+				Name: user.Name,
+				Key: &resourcepb.ResourceKey{
+					Name:      user.Name,
+					Namespace: key.Namespace,
+					Group:     key.Group,
+					Resource:  key.Resource,
+				},
+				Title:  user.Name,
+				Fields: map[string]any{search.USER_LOGIN: user.Spec.Login, search.USER_EMAIL: user.Spec.Email},
+			},
+		})
+	}
+	req := &resource.BulkIndexRequest{Items: items}
+	require.NoError(t, index.BulkIndex(req))
+}
+
+func checkUserSearchQuery(t *testing.T, index resource.ResourceIndex, query *resourcepb.ResourceSearchRequest, orderedExpectedNames []string) {
+	t.Helper()
+	res, err := index.Search(context.Background(), nil, query, nil)
+	require.NoError(t, err)
+	require.Equal(t, int64(len(orderedExpectedNames)), res.TotalHits)
+	names := make([]string, len(res.Results.Rows))
+	for ix, row := range res.Results.Rows {
+		names[ix] = row.Key.Name
+	}
+	assert.ElementsMatch(t, orderedExpectedNames, names)
+}
+
+func newTestUserQueryWithReqs(key resource.NamespacedResource, filterReqs []*resourcepb.Requirement) *resourcepb.ResourceSearchRequest {
+	return &resourcepb.ResourceSearchRequest{
+		Options: &resourcepb.ListOptions{
+			Key: &resourcepb.ResourceKey{
+				Namespace: key.Namespace,
+				Group:     key.Group,
+				Resource:  key.Resource,
+			},
+			Fields: filterReqs,
+		},
+		Limit: 100,
+	}
+}
+
+func newTestsUserQueryWithTitle(key resource.NamespacedResource, title string) *resourcepb.ResourceSearchRequest {
+	return &resourcepb.ResourceSearchRequest{
+		Options: &resourcepb.ListOptions{
+			Key: &resourcepb.ResourceKey{
+				Namespace: key.Namespace,
+				Group:     key.Group,
+				Resource:  key.Resource,
+			},
+		},
+		Query: title,
+		Limit: 100,
+	}
+}

pkg/tests/apis/iam/testdata/user-test-create-duplicate-email-other.yaml

@@ -0,0 +1,11 @@
+apiVersion: iam.grafana.app/v0alpha1
+kind: User
+metadata:
+  namespace: default
+  name: testuser-email-2
+spec:
+  email: testuser-email-1@example
+  login: testuser-email-2
+  name: Test User Email 2
+  provisioned: false
+  role: None

pkg/tests/apis/iam/testdata/user-test-create-duplicate-email-v0.yaml

@@ -0,0 +1,11 @@
+apiVersion: iam.grafana.app/v0alpha1
+kind: User
+metadata:
+  namespace: default
+  name: testuser-email-1
+spec:
+  email: testuser-email-1@example
+  login: testuser-email-1
+  name: Test User Email 1
+  provisioned: false
+  role: None

pkg/tests/apis/iam/testdata/user-test-create-duplicate-login-other.yaml

@@ -0,0 +1,11 @@
+apiVersion: iam.grafana.app/v0alpha1
+kind: User
+metadata:
+  namespace: default
+  name: testuser-login-2
+spec:
+  email: testuser-login-2@example.com
+  login: testuser-login-1
+  name: Test User Login 2
+  provisioned: false
+  role: None

pkg/tests/apis/iam/testdata/user-test-create-duplicate-login-v0.yaml

@@ -0,0 +1,11 @@
+apiVersion: iam.grafana.app/v0alpha1
+kind: User
+metadata:
+  namespace: default
+  name: testuser-login-1
+spec:
+  email: testuser-login-1@example.com
+  login: testuser-login-1
+  name: Test User Login 1
+  provisioned: false
+  role: None

pkg/tests/apis/iam/testdata/user-test-create-v0.yaml

@@ -4,7 +4,7 @@ metadata:
   namespace: default
   name: abcdefghijkl
 spec:
-  email: testuser1@example123.com
+  email: testuser1@example123
   login: testuser1
   name: Test User 1
   provisioned: false

pkg/tests/apis/iam/testdata/user-test-create-v1.yaml

@@ -0,0 +1,11 @@
+apiVersion: iam.grafana.app/v0alpha1
+kind: User
+metadata:
+  namespace: default
+  name: testuser2
+spec:
+  name: Test User 2
+  login: testuser2
+  email: testuser2@example
+  provisioned: false
+  role: Viewer

pkg/tests/apis/iam/user_integration_test.go

@@ -23,7 +23,7 @@ func TestIntegrationUsers(t *testing.T) {
 	// TODO: Figure out why rest.Mode4 is failing
 	modes := []rest.DualWriterMode{rest.Mode0, rest.Mode1, rest.Mode2, rest.Mode3}
 	for _, mode := range modes {
-		t.Run(fmt.Sprintf("User CRUD operations with dual writer mode %d", mode), func(t *testing.T) {
+		t.Run(fmt.Sprintf("DualWriterMode %d", mode), func(t *testing.T) {
 			helper := apis.NewK8sTestHelper(t, testinfra.GrafanaOpts{
 				AppModeProduction:    false,
 				DisableAnonymous:     true,
@@ -36,8 +36,14 @@ func TestIntegrationUsers(t *testing.T) {
 				EnableFeatureToggles: []string{
 					featuremgmt.FlagGrafanaAPIServerWithExperimentalAPIs,
 					featuremgmt.FlagKubernetesAuthnMutation,
+					featuremgmt.FlagUnifiedStorageSearch,
 				},
 			})
+
+			t.Cleanup(func() {
+				helper.Shutdown()
+			})
+
 			doUserCRUDTestsUsingTheNewAPIs(t, helper)
 
 			if mode < 3 {
@@ -64,7 +70,7 @@ func doUserCRUDTestsUsingTheNewAPIs(t *testing.T, helper *apis.K8sTestHelper) {
 
 		// Verify creation response
 		createdSpec := created.Object["spec"].(map[string]interface{})
-		require.Equal(t, "testuser1@example123.com", createdSpec["email"])
+		require.Equal(t, "testuser1@example123", createdSpec["email"])
 		require.Equal(t, "testuser1", createdSpec["login"])
 		require.Equal(t, "Test User 1", createdSpec["name"])
 		require.Equal(t, false, createdSpec["provisioned"])
@@ -82,7 +88,7 @@ func doUserCRUDTestsUsingTheNewAPIs(t *testing.T, helper *apis.K8sTestHelper) {
 
 		// Verify fetched user matches created user
 		fetchedSpec := fetched.Object["spec"].(map[string]interface{})
-		require.Equal(t, "testuser1@example123.com", fetchedSpec["email"])
+		require.Equal(t, "testuser1@example123", fetchedSpec["email"])
 		require.Equal(t, "testuser1", fetchedSpec["login"])
 		require.Equal(t, "Test User 1", fetchedSpec["name"])
 		require.Equal(t, false, fetchedSpec["provisioned"])
@@ -110,7 +116,7 @@ func doUserCRUDTestsUsingTheNewAPIs(t *testing.T, helper *apis.K8sTestHelper) {
 		})
 
 		// Create the user
-		created, err := userClient.Resource.Create(ctx, helper.LoadYAMLOrJSONFile("testdata/user-test-create-2-v0.yaml"), metav1.CreateOptions{})
+		created, err := userClient.Resource.Create(ctx, helper.LoadYAMLOrJSONFile("testdata/user-test-create-v1.yaml"), metav1.CreateOptions{})
 		require.NoError(t, err)
 		require.NotNil(t, created)
 
@@ -122,7 +128,7 @@ func doUserCRUDTestsUsingTheNewAPIs(t *testing.T, helper *apis.K8sTestHelper) {
 		// Modify the user spec
 		spec := userToUpdate.Object["spec"].(map[string]interface{})
 		spec["name"] = "Updated Test User"
-		spec["email"] = "updated.test.user@example.com"
+		spec["email"] = "updated.test.user@example"
 		userToUpdate.Object["spec"] = spec
 
 		// Update the user
@@ -133,14 +139,14 @@ func doUserCRUDTestsUsingTheNewAPIs(t *testing.T, helper *apis.K8sTestHelper) {
 		// Verify the update response
 		updatedSpec := updated.Object["spec"].(map[string]interface{})
 		require.Equal(t, "Updated Test User", updatedSpec["name"])
-		require.Equal(t, "updated.test.user@example.com", updatedSpec["email"])
+		require.Equal(t, "updated.test.user@example", updatedSpec["email"])
 
 		// Fetch again to confirm
 		fetched, err := userClient.Resource.Get(ctx, createdUID, metav1.GetOptions{})
 		require.NoError(t, err)
 		fetchedSpec := fetched.Object["spec"].(map[string]interface{})
 		require.Equal(t, "Updated Test User", fetchedSpec["name"])
-		require.Equal(t, "updated.test.user@example.com", fetchedSpec["email"])
+		require.Equal(t, "updated.test.user@example", fetchedSpec["email"])
 
 		// Cleanup
 		err = userClient.Resource.Delete(ctx, fetched.GetName(), metav1.DeleteOptions{})
@@ -170,6 +176,144 @@ func doUserCRUDTestsUsingTheNewAPIs(t *testing.T, helper *apis.K8sTestHelper) {
 			})
 		}
 	})
+
+	t.Run("should not be able to create a user with a duplicate email", func(t *testing.T) {
+		ctx := context.Background()
+		userClient := helper.GetResourceClient(apis.ResourceClientArgs{
+			User:      helper.Org1.Admin,
+			Namespace: helper.Namespacer(helper.Org1.Admin.Identity.GetOrgID()),
+			GVR:       gvrUsers,
+		})
+
+		// Create the first user
+		created, err := userClient.Resource.Create(ctx, helper.LoadYAMLOrJSONFile("testdata/user-test-create-duplicate-email-v0.yaml"), metav1.CreateOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, created)
+
+		// Attempt to create another user with the same email
+		_, err = userClient.Resource.Create(ctx, helper.LoadYAMLOrJSONFile("testdata/user-test-create-duplicate-email-other.yaml"), metav1.CreateOptions{})
+		require.Error(t, err)
+		var statusErr *errors.StatusError
+		require.ErrorAs(t, err, &statusErr)
+		require.Equal(t, int32(409), statusErr.ErrStatus.Code)
+		require.Contains(t, statusErr.ErrStatus.Message, "email 'testuser-email-1@example' is already taken")
+
+		// Cleanup
+		err = userClient.Resource.Delete(ctx, created.GetName(), metav1.DeleteOptions{})
+		require.NoError(t, err)
+	})
+
+	t.Run("should not be able to create a user with a duplicate login", func(t *testing.T) {
+		ctx := context.Background()
+		userClient := helper.GetResourceClient(apis.ResourceClientArgs{
+			User:      helper.Org1.Admin,
+			Namespace: helper.Namespacer(helper.Org1.Admin.Identity.GetOrgID()),
+			GVR:       gvrUsers,
+		})
+
+		// Create the first user
+		created, err := userClient.Resource.Create(ctx, helper.LoadYAMLOrJSONFile("testdata/user-test-create-duplicate-login-v0.yaml"), metav1.CreateOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, created)
+
+		// Attempt to create a second user with the same login
+		_, err = userClient.Resource.Create(ctx, helper.LoadYAMLOrJSONFile("testdata/user-test-create-duplicate-login-other.yaml"), metav1.CreateOptions{})
+		require.Error(t, err)
+		var statusErr *errors.StatusError
+		require.ErrorAs(t, err, &statusErr)
+		require.Equal(t, int32(409), statusErr.ErrStatus.Code)
+		require.Contains(t, statusErr.ErrStatus.Message, "login 'testuser-login-1' is already taken")
+
+		// Cleanup
+		err = userClient.Resource.Delete(ctx, created.GetName(), metav1.DeleteOptions{})
+		require.NoError(t, err)
+	})
+
+	t.Run("should not be able to update a user with an existing email", func(t *testing.T) {
+		ctx := context.Background()
+		userClient := helper.GetResourceClient(apis.ResourceClientArgs{
+			User:      helper.Org1.Admin,
+			Namespace: helper.Namespacer(helper.Org1.Admin.Identity.GetOrgID()),
+			GVR:       gvrUsers,
+		})
+
+		// Create the first user
+		user1, err := userClient.Resource.Create(ctx, helper.LoadYAMLOrJSONFile("testdata/user-test-create-v0.yaml"), metav1.CreateOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, user1)
+
+		// Create the second user
+		user2, err := userClient.Resource.Create(ctx, helper.LoadYAMLOrJSONFile("testdata/user-test-create-v1.yaml"), metav1.CreateOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, user2)
+
+		// Get the user to update
+		userToUpdate, err := userClient.Resource.Get(ctx, user2.GetName(), metav1.GetOptions{})
+		require.NoError(t, err)
+
+		// Modify the user spec to have the same email as user1
+		spec := userToUpdate.Object["spec"].(map[string]interface{})
+		user1Spec := user1.Object["spec"].(map[string]interface{})
+		spec["email"] = user1Spec["email"]
+		userToUpdate.Object["spec"] = spec
+
+		// Attempt to update the user
+		_, err = userClient.Resource.Update(ctx, userToUpdate, metav1.UpdateOptions{})
+		require.Error(t, err)
+		var statusErr *errors.StatusError
+		require.ErrorAs(t, err, &statusErr)
+		require.Equal(t, int32(409), statusErr.ErrStatus.Code)
+		require.Contains(t, statusErr.ErrStatus.Message, "email 'testuser1@example123' is already taken")
+
+		// Cleanup
+		err = userClient.Resource.Delete(ctx, user1.GetName(), metav1.DeleteOptions{})
+		require.NoError(t, err)
+		err = userClient.Resource.Delete(ctx, user2.GetName(), metav1.DeleteOptions{})
+		require.NoError(t, err)
+	})
+
+	t.Run("should not be able to update a user with an existing login", func(t *testing.T) {
+		ctx := context.Background()
+		userClient := helper.GetResourceClient(apis.ResourceClientArgs{
+			User:      helper.Org1.Admin,
+			Namespace: helper.Namespacer(helper.Org1.Admin.Identity.GetOrgID()),
+			GVR:       gvrUsers,
+		})
+
+		// Create the first user
+		user1, err := userClient.Resource.Create(ctx, helper.LoadYAMLOrJSONFile("testdata/user-test-create-v0.yaml"), metav1.CreateOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, user1)
+
+		// Create the second user
+		user2, err := userClient.Resource.Create(ctx, helper.LoadYAMLOrJSONFile("testdata/user-test-create-v1.yaml"), metav1.CreateOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, user2)
+
+		// Get the user to update
+		userToUpdate, err := userClient.Resource.Get(ctx, user2.GetName(), metav1.GetOptions{})
+		require.NoError(t, err)
+
+		// Modify the user spec to have the same login as user1
+		spec := userToUpdate.Object["spec"].(map[string]interface{})
+		user1Spec := user1.Object["spec"].(map[string]interface{})
+		spec["login"] = user1Spec["login"]
+		userToUpdate.Object["spec"] = spec
+
+		// Attempt to update the user
+		_, err = userClient.Resource.Update(ctx, userToUpdate, metav1.UpdateOptions{})
+		require.Error(t, err)
+		var statusErr *errors.StatusError
+		require.ErrorAs(t, err, &statusErr)
+		require.Equal(t, int32(409), statusErr.ErrStatus.Code)
+		require.Contains(t, statusErr.ErrStatus.Message, "login 'testuser1' is already taken")
+
+		// Cleanup
+		err = userClient.Resource.Delete(ctx, user1.GetName(), metav1.DeleteOptions{})
+		require.NoError(t, err)
+		err = userClient.Resource.Delete(ctx, user2.GetName(), metav1.DeleteOptions{})
+		require.NoError(t, err)
+	})
 }
 
 func doUserCRUDTestsUsingTheLegacyAPIs(t *testing.T, helper *apis.K8sTestHelper) {
@@ -182,7 +326,7 @@ func doUserCRUDTestsUsingTheLegacyAPIs(t *testing.T, helper *apis.K8sTestHelper)
 
 		legacyUserPayload := `{
 			"name": "Legacy User 3",
-			"email": "legacyuser3@example.com",
+			"email": "legacyuser3@example",
 			"login": "legacyuser3",
 			"password": "password123"
 		}`
@@ -205,7 +349,7 @@ func doUserCRUDTestsUsingTheLegacyAPIs(t *testing.T, helper *apis.K8sTestHelper)
 
 		// Verify fetched user matches created user
 		userSpec := user.Object["spec"].(map[string]interface{})
-		require.Equal(t, "legacyuser3@example.com", userSpec["email"])
+		require.Equal(t, "legacyuser3@example", userSpec["email"])
 		require.Equal(t, "legacyuser3", userSpec["login"])
 		require.Equal(t, "Legacy User 3", userSpec["name"])
 		require.Equal(t, false, userSpec["provisioned"])