SetAssociatedAccounts, DisassociateAccounts, AssociateAccounts with support for multiple accounts. (#439)

internal/authtoken/repository_test.go

@@ -427,7 +427,7 @@ func TestRepository_ValidateToken_expired(t *testing.T) {
 	aAcct := allocAuthAccount()
 	aAcct.PublicId = baseAT.GetAuthAccountId()
 	require.NoError(t, rw.LookupByPublicId(context.Background(), aAcct))
-	iamUser, err := iamRepo.LookupUser(context.Background(), aAcct.GetIamUserId())
+	iamUser, _, err := iamRepo.LookupUser(context.Background(), aAcct.GetIamUserId())
 	require.NoError(t, err)
 	require.NotNil(t, iamUser)
 

internal/cmd/base/initial_resources.go

@@ -117,10 +117,9 @@ func (b *Server) CreateInitialAuthMethod(ctx context.Context) (*password.AuthMet
 	if u, err = iamRepo.CreateUser(cancelCtx, u, opts...); err != nil {
 		return nil, nil, fmt.Errorf("error creating initial admin user: %w", err)
 	}
-	if u, err = iamRepo.AssociateUserWithAccount(cancelCtx, u.GetPublicId(), acct.GetPublicId()); err != nil {
+	if _, err = iamRepo.AssociateAccounts(cancelCtx, u.GetPublicId(), u.GetVersion(), []string{acct.GetPublicId()}); err != nil {
 		return nil, nil, fmt.Errorf("error associating initial admin user with account: %w", err)
 	}
-
 	// Create a role tying them together
 	pr, err := iam.NewRole(scope.Global.String(),
 		iam.WithName("Generated global scope admin role"),

internal/db/read_writer.go

@@ -46,6 +46,12 @@ type Reader interface {
 	// default limits are used for results.
 	SearchWhere(ctx context.Context, resources interface{}, where string, args []interface{}, opt ...Option) error
 
+	// Query will run the raw query and return the *sql.Rows results. Query will
+	// operate within the context of any ongoing transaction for the db.Reader.  The
+	// caller must close the returned *sql.Rows. Query can/should be used in
+	// combination with ScanRows.
+	Query(sql string, values []interface{}, opt ...Option) (*sql.Rows, error)
+
 	// ScanRows will scan sql rows into the interface provided
 	ScanRows(rows *sql.Rows, result interface{}) error
 
@@ -206,6 +212,21 @@ func (rw *Db) Exec(sql string, values []interface{}, opt ...Option) (int, error)
 	return int(gormDb.RowsAffected), nil
 }
 
+// Query will run the raw query and return the *sql.Rows results. Query will
+// operate within the context of any ongoing transaction for the db.Reader.  The
+// caller must close the returned *sql.Rows. Query can/should be used in
+// combination with ScanRows.
+func (rw *Db) Query(sql string, values []interface{}, opt ...Option) (*sql.Rows, error) {
+	if sql == "" {
+		return nil, fmt.Errorf("raw missing sql: %w", ErrInvalidParameter)
+	}
+	gormDb := rw.underlying.Raw(sql, values...)
+	if gormDb.Error != nil {
+		return nil, fmt.Errorf("exec: failed: %w", gormDb.Error)
+	}
+	return gormDb.Rows()
+}
+
 // Scan rows will scan the rows into the interface
 func (rw *Db) ScanRows(rows *sql.Rows, result interface{}) error {
 	if rw.underlying == nil {

internal/db/read_writer_test.go

@@ -1525,6 +1525,35 @@ func TestDb_ScanRows(t *testing.T) {
 	})
 }
 
+func TestDb_Query(t *testing.T) {
+	t.Parallel()
+	db, _ := TestSetup(t, "postgres")
+	t.Run("valid", func(t *testing.T) {
+		assert, require := assert.New(t), require.New(t)
+		rw := Db{underlying: db}
+		user, err := db_test.NewTestUser()
+		user.Name = "alice"
+		require.NoError(err)
+		err = rw.Create(context.Background(), user)
+		require.NoError(err)
+		assert.NotEmpty(user.Id)
+		assert.Equal("alice", user.Name)
+
+		where := "select * from db_test_user where name in ($1, $2)"
+		rows, err := rw.Query(where, []interface{}{"alice", "bob"})
+		require.NoError(err)
+		defer func() { err := rows.Close(); assert.NoError(err) }()
+		for rows.Next() {
+			u, err := db_test.NewTestUser()
+			require.NoError(err)
+			// scan the row into your Gorm struct
+			err = rw.ScanRows(rows, &u)
+			require.NoError(err)
+			assert.Equal(user.PublicId, u.PublicId)
+		}
+	})
+}
+
 func TestDb_CreateItems(t *testing.T) {
 	db, _ := TestSetup(t, "postgres")
 	testOplogResourceId := testId(t)

internal/iam/query.go

@@ -19,6 +19,49 @@ const (
 	whereValidAuthMethod = `select count(*) from auth_method where public_id = $1 and scope_id = $2`
 
 	// insertAuthMethod - insert a row directly into auth_method (TODO - this
-	// should be replaced with calls to the auth method repo)
+	// should be replaced with calls to the auth method repo).
 	insertAuthMethod = `insert into auth_method (public_id, scope_id) values ($1, $2)`
+
+	accountChangesQuery = `
+	with
+	final_accounts (account_id) as (
+	  -- returns the SET list
+	  select public_id
+		from auth_account
+	   where public_id in (%s)
+	),
+	current_accounts (account_id) as (
+	  -- returns the current list
+	  select public_id
+		from auth_account
+	   where iam_user_id = $1
+	),
+	keep_accounts (account_id) as (
+	  -- returns the KEEP list
+	  select account_id
+		from current_accounts
+	   where account_id in (select * from final_accounts)
+	),
+	delete_accounts (account_id) as (
+	  -- returns the DELETE list
+	  select account_id
+		from current_accounts
+	   where account_id not in (select * from final_accounts)
+	),
+	insert_accounts (account_id) as (
+	  -- returns the ADD list
+	  select account_id
+		from final_accounts
+	   where account_id not in (select * from keep_accounts)
+	),
+	final (action, account_id) as (
+	  select 'disassociate', account_id
+		from delete_accounts
+	   union
+	  select 'associate', account_id
+		from insert_accounts
+	)
+	select * from final
+	order by action, account_id;
+	`
 )

internal/iam/repository_group_test.go

@@ -840,7 +840,7 @@ func TestRepository_AddGroupMembers(t *testing.T) {
 			}
 			for _, id := range tt.args.userIds {
 				assert.NotEmpty(gotMembers[id])
-				u, err := repo.LookupUser(context.Background(), id)
+				u, _, err := repo.LookupUser(context.Background(), id)
 				assert.NoError(err)
 				assert.Equal(id, u.PublicId)
 			}

internal/iam/repository_user.go

@@ -48,24 +48,25 @@ func (r *Repository) CreateUser(ctx context.Context, user *User, opt ...Option)
 	return resource.(*User), err
 }
 
-// UpdateUser will update a user in the repository and return the written user.
-// fieldMaskPaths provides field_mask.proto paths for fields that should
-// be updated.  Fields will be set to NULL if the field is a zero value and
-// included in fieldMask. Name and Description are the only updatable fields,
-// If no updatable fields are included in the fieldMaskPaths, then an error is returned.
-func (r *Repository) UpdateUser(ctx context.Context, user *User, version uint32, fieldMaskPaths []string, opt ...Option) (*User, int, error) {
+// UpdateUser will update a user in the repository and return the written user
+// plus its associated account ids. fieldMaskPaths provides field_mask.proto
+// paths for fields that should be updated.  Fields will be set to NULL if the
+// field is a zero value and included in fieldMask. Name and Description are the
+// only updatable fields, if no updatable fields are included in the
+// fieldMaskPaths, then an error is returned.
+func (r *Repository) UpdateUser(ctx context.Context, user *User, version uint32, fieldMaskPaths []string, opt ...Option) (*User, []string, int, error) {
 	if user == nil {
-		return nil, db.NoRowsAffected, fmt.Errorf("update user: missing user %w", db.ErrInvalidParameter)
+		return nil, nil, db.NoRowsAffected, fmt.Errorf("update user: missing user %w", db.ErrInvalidParameter)
 	}
 	if user.PublicId == "" {
-		return nil, db.NoRowsAffected, fmt.Errorf("update user: missing user public id %w", db.ErrInvalidParameter)
+		return nil, nil, db.NoRowsAffected, fmt.Errorf("update user: missing user public id %w", db.ErrInvalidParameter)
 	}
 	for _, f := range fieldMaskPaths {
 		switch {
 		case strings.EqualFold("name", f):
 		case strings.EqualFold("description", f):
 		default:
-			return nil, db.NoRowsAffected, fmt.Errorf("update user: field: %s: %w", f, db.ErrInvalidFieldMask)
+			return nil, nil, db.NoRowsAffected, fmt.Errorf("update user: field: %s: %w", f, db.ErrInvalidFieldMask)
 		}
 	}
 	var dbMask, nullFields []string
@@ -78,36 +79,102 @@ func (r *Repository) UpdateUser(ctx context.Context, user *User, version uint32,
 		nil,
 	)
 	if len(dbMask) == 0 && len(nullFields) == 0 {
-		return nil, db.NoRowsAffected, fmt.Errorf("update user: %w", db.ErrEmptyFieldMask)
+		return nil, nil, db.NoRowsAffected, fmt.Errorf("update user: %w", db.ErrEmptyFieldMask)
 	}
 
-	u := user.Clone()
-	resource, rowsUpdated, err := r.update(ctx, u.(*User), version, dbMask, nullFields, opt...)
+	u := user.Clone().(*User)
+	metadata, err := r.stdMetadata(ctx, u)
+	if err != nil {
+		return nil, nil, db.NoRowsAffected, fmt.Errorf("update user: error getting metadata for update: %w", err)
+	}
+	metadata["op-type"] = []string{oplog.OpType_OP_TYPE_UPDATE.String()}
+
+	dbOpts := []db.Option{
+		db.WithVersion(&version),
+	}
+	opts := getOpts(opt...)
+	if opts.withSkipVetForWrite {
+		dbOpts = append(dbOpts, db.WithSkipVetForWrite(true))
+	}
+
+	scope, err := u.GetScope(ctx, r.reader)
+	if err != nil {
+		return nil, nil, db.NoRowsAffected, fmt.Errorf("update user: unable to get scope: %w", err)
+	}
+
+	oplogWrapper, err := r.kms.GetWrapper(ctx, scope.GetPublicId(), kms.KeyPurposeOplog)
+	if err != nil {
+		return nil, nil, db.NoRowsAffected, fmt.Errorf("update user: unable to get oplog wrapper: %w", err)
+	}
+	dbOpts = append(dbOpts, db.WithOplog(oplogWrapper, metadata))
+
+	var rowsUpdated int
+	var returnedUser *User
+	var currentAccountIds []string
+	_, err = r.writer.DoTx(
+		ctx,
+		db.StdRetryCnt,
+		db.ExpBackoff{},
+		func(reader db.Reader, w db.Writer) error {
+			returnedUser = u.Clone().(*User)
+			rowsUpdated, err = w.Update(
+				ctx,
+				returnedUser,
+				dbMask,
+				nullFields,
+				dbOpts...,
+			)
+			if err == nil && rowsUpdated > 1 {
+				// return err, which will result in a rollback of the update
+				return errors.New("error more than 1 resource would have been updated ")
+			}
+			if err != nil {
+				return err
+			}
+			// we need a new repo, that's using the same reader/writer as this TxHandler
+			txRepo := &Repository{
+				reader: reader,
+				writer: w,
+				kms:    r.kms,
+				// intentionally not setting the defaultLimit, so we'll get all
+				// the account ids without a limit
+			}
+			currentAccountIds, err = txRepo.ListAssociatedAccountIds(ctx, user.PublicId)
+			if err != nil {
+				return fmt.Errorf("unable to retrieve current account ids after update: %w", err)
+			}
+			return nil
+		},
+	)
 	if err != nil {
 		if db.IsUniqueError(err) {
-			return nil, db.NoRowsAffected, fmt.Errorf("update user: user %s already exists in org %s", user.Name, user.ScopeId)
+			return nil, nil, db.NoRowsAffected, fmt.Errorf("update user: user %s already exists in org %s", user.Name, user.ScopeId)
 		}
-		return nil, db.NoRowsAffected, fmt.Errorf("update user: %w for %s", err, user.PublicId)
+		return nil, nil, db.NoRowsAffected, fmt.Errorf("update user: %w for %s", err, user.PublicId)
 	}
-	return resource.(*User), rowsUpdated, err
+	return returnedUser, currentAccountIds, rowsUpdated, err
 }
 
-// LookupUser will look up a user in the repository.  If the user is not
-// found, it will return nil, nil.
-func (r *Repository) LookupUser(ctx context.Context, withPublicId string, opt ...Option) (*User, error) {
-	if withPublicId == "" {
-		return nil, fmt.Errorf("lookup user: missing public id %w", db.ErrInvalidParameter)
+// LookupUser will look up a user and its associated account ids in the
+// repository.  If the user is not found, it will return nil, nil, nil.
+func (r *Repository) LookupUser(ctx context.Context, userId string, opt ...Option) (*User, []string, error) {
+	if userId == "" {
+		return nil, nil, fmt.Errorf("lookup user: missing public id %w", db.ErrInvalidParameter)
 	}
 
 	user := allocUser()
-	user.PublicId = withPublicId
+	user.PublicId = userId
 	if err := r.reader.LookupByPublicId(ctx, &user); err != nil {
 		if errors.Is(err, db.ErrRecordNotFound) {
-			return nil, nil
+			return nil, nil, nil
 		}
-		return nil, fmt.Errorf("lookup user: failed %w for %s", err, withPublicId)
+		return nil, nil, fmt.Errorf("lookup user: failed %w for %s", err, userId)
 	}
-	return &user, nil
+	currentAccountIds, err := r.ListAssociatedAccountIds(ctx, userId)
+	if err != nil {
+		return nil, nil, fmt.Errorf("lookup user: unable to retrieve current account ids: %w", err)
+	}
+	return &user, currentAccountIds, nil
 }
 
 // DeleteUser will delete a user from the repository
@@ -264,158 +331,469 @@ func (r *Repository) getUserWithAccount(ctx context.Context, withAccountId strin
 	return &u, nil
 }
 
-// AssociateUserWithAccount will associate a user with an existing account.
-// The account must not already be associated with a different user.  No
-// options are currently supported.
-func (r *Repository) AssociateUserWithAccount(ctx context.Context, userPublicId, accountId string, opt ...Option) (*User, error) {
-	opts := getOpts(opt...)
-	if userPublicId == "" {
-		return nil, fmt.Errorf("associate user with account: missing user public id %w", db.ErrInvalidParameter)
+// ListAssociatedAccountIds returns the account ids for the userId and supports the
+// WithLimit option. Returns nil, nil when no associated accounts are found.
+func (r *Repository) ListAssociatedAccountIds(ctx context.Context, userId string, opt ...Option) ([]string, error) {
+	if userId == "" {
+		return nil, fmt.Errorf("list auth account ids: missing user id: %w", db.ErrInvalidParameter)
 	}
-	if accountId == "" {
-		return nil, fmt.Errorf("associate user with account: missing account id %w", db.ErrInvalidParameter)
+	var accounts []*authAccount
+	if err := r.list(ctx, &accounts, "iam_user_id = ?", []interface{}{userId}, opt...); err != nil {
+		return nil, fmt.Errorf("list auth account ids: %w", err)
+	}
+	if len(accounts) == 0 {
+		return nil, nil
+	}
+	ids := make([]string, 0, len(accounts))
+	for _, aa := range accounts {
+		ids = append(ids, aa.PublicId)
+	}
+	return ids, nil
+}
+
+// AssociateAccounts will associate a user with existing accounts and
+// return a list of all associated account ids for the user. The accounts must
+// not already be associated with different users.  No options are currently
+// supported.
+func (r *Repository) AssociateAccounts(ctx context.Context, userId string, userVersion uint32, accountIds []string, opt ...Option) ([]string, error) {
+	if userId == "" {
+		return nil, fmt.Errorf("associate accounts: missing user public id %w", db.ErrInvalidParameter)
+	}
+	if userVersion == 0 {
+		return nil, fmt.Errorf("associate accounts: missing user version %w", db.ErrInvalidParameter)
+	}
+	if len(accountIds) == 0 {
+		return nil, fmt.Errorf("associate accounts: missing account id %w", db.ErrInvalidParameter)
 	}
 
 	user := allocUser()
-	user.PublicId = userPublicId
+	user.PublicId = userId
 
 	err := r.reader.LookupById(ctx, &user)
 	if err != nil {
-		return nil, fmt.Errorf("associate user with account: unable to lookup user %s: %w", userPublicId, err)
+		return nil, fmt.Errorf("associate accounts: unable to lookup user %s: %w", userId, err)
 	}
-
-	acct := allocAccount()
-	acct.PublicId = accountId
-	err = r.reader.LookupByPublicId(context.Background(), &acct)
+	oplogWrapper, err := r.kms.GetWrapper(ctx, user.ScopeId, kms.KeyPurposeOplog)
 	if err != nil {
-		return nil, fmt.Errorf("associate user with account: unable to lookup account %s: %w", accountId, err)
-	}
-	// first, let's handle the case where the account is already
-	// associated with the user, so we're done!
-	if acct.IamUserId == userPublicId {
-		return &user, nil
+		return nil, fmt.Errorf("associate accounts: unable to get oplog wrapper: %w", err)
 	}
 
-	if !opts.withDisassociate {
-		if acct.IamUserId != "" && acct.IamUserId != userPublicId {
-			return nil, fmt.Errorf("associate user with account: %s account is already associated with a user: %w", accountId, db.ErrInvalidParameter)
+	var currentAccountIds []string
+	_, err = r.writer.DoTx(
+		ctx,
+		db.StdRetryCnt,
+		db.ExpBackoff{},
+		func(reader db.Reader, w db.Writer) error {
+			userTicket, err := w.GetTicket(&user)
+			if err != nil {
+				return fmt.Errorf("associate accounts: unable to get ticket: %w", err)
+			}
+			updatedUser := allocUser()
+			updatedUser.PublicId = userId
+			updatedUser.Version = userVersion + 1
+			var userOplogMsg oplog.Message
+			rowsUpdated, err := w.Update(ctx, &updatedUser, []string{"Version"}, nil, db.NewOplogMsg(&userOplogMsg), db.WithVersion(&userVersion))
+			if err != nil {
+				return fmt.Errorf("associate accounts: unable to update user version: %w", err)
+			}
+			if rowsUpdated != 1 {
+				return fmt.Errorf("associate accounts: updated user and %d rows updated", rowsUpdated)
+			}
+			if err := associateUserWithAccounts(ctx, r.kms, reader, w, user.PublicId, accountIds); err != nil {
+				return fmt.Errorf("associate accounts: %w", err)
+			}
+			metadata := oplog.Metadata{
+				"op-type":            []string{oplog.OpType_OP_TYPE_UPDATE.String()},
+				"scope-id":           []string{user.ScopeId},
+				"scope-type":         []string{scope.Org.String()},
+				"resource-public-id": []string{user.PublicId},
+			}
+			if err := w.WriteOplogEntryWith(ctx, oplogWrapper, userTicket, metadata, []*oplog.Message{&userOplogMsg}); err != nil {
+				return fmt.Errorf("associate accounts: unable to write oplog: %w", err)
+			}
+			// we need a new repo, that's using the same reader/writer as this TxHandler
+			txRepo := &Repository{
+				reader: reader,
+				writer: w,
+				kms:    r.kms,
+				// intentionally not setting the defaultLimit, so we'll get all
+				// the account ids without a limit
+			}
+			currentAccountIds, err = txRepo.ListAssociatedAccountIds(ctx, user.PublicId)
+			if err != nil {
+				return fmt.Errorf("associate accounts: unable to retrieve current account ids after adds: %w", err)
+			}
+			return nil
+		},
+	)
+	if err != nil {
+		return nil, fmt.Errorf("associate accounts: error associating account ids: %w", err)
+	}
+	return currentAccountIds, nil
+}
+
+// DisassociateAccounts will disassociate a user from existing accounts and
+// return a list of all associated account ids for the user. The accounts must
+// not be associated with different users.  No options are currently
+// supported.
+func (r *Repository) DisassociateAccounts(ctx context.Context, userId string, userVersion uint32, accountIds []string, opt ...Option) ([]string, error) {
+	if userId == "" {
+		return nil, fmt.Errorf("disassociate accounts: missing user public id %w", db.ErrInvalidParameter)
+	}
+	if userVersion == 0 {
+		return nil, fmt.Errorf("disassociate accounts: missing user version %w", db.ErrInvalidParameter)
+	}
+	if len(accountIds) == 0 {
+		return nil, fmt.Errorf("disassociate accounts: missing account id %w", db.ErrInvalidParameter)
+	}
+
+	user := allocUser()
+	user.PublicId = userId
+	err := r.reader.LookupById(ctx, &user)
+	if err != nil {
+		return nil, fmt.Errorf("disassociate accounts: unable to lookup user %s: %w", userId, err)
+	}
+	oplogWrapper, err := r.kms.GetWrapper(ctx, user.ScopeId, kms.KeyPurposeOplog)
+	if err != nil {
+		return nil, fmt.Errorf("disassociate accounts: unable to get oplog wrapper: %w", err)
+	}
+
+	var currentAccountIds []string
+	_, err = r.writer.DoTx(
+		ctx,
+		db.StdRetryCnt,
+		db.ExpBackoff{},
+		func(reader db.Reader, w db.Writer) error {
+			userTicket, err := w.GetTicket(&user)
+			if err != nil {
+				return fmt.Errorf("disassociate accounts: unable to get ticket: %w", err)
+			}
+			updatedUser := allocUser()
+			updatedUser.PublicId = userId
+			updatedUser.Version = userVersion + 1
+			var userOplogMsg oplog.Message
+			rowsUpdated, err := w.Update(ctx, &updatedUser, []string{"Version"}, nil, db.NewOplogMsg(&userOplogMsg), db.WithVersion(&userVersion))
+			if err != nil {
+				return fmt.Errorf("disassociate accounts: unable to update user version: %w", err)
+			}
+			if rowsUpdated != 1 {
+				return fmt.Errorf("disassociate accounts: updated user and %d rows updated", rowsUpdated)
+			}
+			if err := dissociateUserFromAccounts(ctx, r.kms, reader, w, user.PublicId, accountIds); err != nil {
+				return fmt.Errorf("disassociate accounts: %w", err)
+			}
+			metadata := oplog.Metadata{
+				"op-type":            []string{oplog.OpType_OP_TYPE_UPDATE.String()},
+				"scope-id":           []string{user.ScopeId},
+				"scope-type":         []string{scope.Org.String()},
+				"resource-public-id": []string{user.PublicId},
+			}
+			if err := w.WriteOplogEntryWith(ctx, oplogWrapper, userTicket, metadata, []*oplog.Message{&userOplogMsg}); err != nil {
+				return fmt.Errorf("disassociate accounts: unable to write oplog: %w", err)
+			}
+			// we need a new repo, that's using the same reader/writer as this TxHandler
+			txRepo := &Repository{
+				reader: reader,
+				writer: w,
+				kms:    r.kms,
+				// intentionally not setting the defaultLimit, so we'll get all
+				// the account ids without a limit
+			}
+			currentAccountIds, err = txRepo.ListAssociatedAccountIds(ctx, user.PublicId)
+			if err != nil {
+				return fmt.Errorf("disassociate accounts: unable to retrieve current account ids after adds: %w", err)
+			}
+			return nil
+		},
+	)
+	if err != nil {
+		return nil, fmt.Errorf("disassociate accounts: error associating account ids: %w", err)
+	}
+	return currentAccountIds, nil
+}
+
+// DisassociatedAccounts will associate a user with existing accounts and
+// return a list of all associated account ids for the user. The accounts must
+// not already be associated with different users.  No options are currently
+// supported.
+func (r *Repository) SetAssociatedAccounts(ctx context.Context, userId string, userVersion uint32, accountIds []string, opt ...Option) ([]string, error) {
+	if userId == "" {
+		return nil, fmt.Errorf("set associated accounts: missing user public id %w", db.ErrInvalidParameter)
+	}
+	if userVersion == 0 {
+		return nil, fmt.Errorf("set associated accounts: missing user version %w", db.ErrInvalidParameter)
+	}
+
+	user := allocUser()
+	user.PublicId = userId
+	err := r.reader.LookupById(ctx, &user)
+	if err != nil {
+		return nil, fmt.Errorf("set associated accounts: unable to lookup user %s: %w", userId, err)
+	}
+	oplogWrapper, err := r.kms.GetWrapper(ctx, user.ScopeId, kms.KeyPurposeOplog)
+	if err != nil {
+		return nil, fmt.Errorf("set associated accounts: unable to get oplog wrapper: %w", err)
+	}
+	var currentAccountIds []string
+	_, err = r.writer.DoTx(
+		ctx,
+		db.StdRetryCnt,
+		db.ExpBackoff{},
+		func(reader db.Reader, w db.Writer) error {
+			associateIds, disassociateIds, err := associationChanges(ctx, reader, userId, accountIds)
+			if err != nil {
+				return fmt.Errorf("set associated accounts: unable to determine changes: %w", err)
+			}
+
+			// just in case we've got nothing to do...
+			if len(associateIds) == 0 && len(disassociateIds) == 0 {
+				// we need a new repo, that's using the same reader/writer as this TxHandler
+				txRepo := &Repository{
+					reader: reader,
+					writer: w,
+					kms:    r.kms,
+					// intentionally not setting the defaultLimit, so we'll get all
+					// the account ids without a limit
+				}
+				currentAccountIds, err = txRepo.ListAssociatedAccountIds(ctx, userId)
+				if err != nil {
+					return fmt.Errorf("set associated accounts: unable to retrieve current account ids after set: %w", err)
+				}
+				return nil
+			}
+			userTicket, err := w.GetTicket(&user)
+			if err != nil {
+				return fmt.Errorf("set associated accounts: unable to get ticket: %w", err)
+			}
+			updatedUser := allocUser()
+			updatedUser.PublicId = userId
+			updatedUser.Version = userVersion + 1
+			var userOplogMsg oplog.Message
+			rowsUpdated, err := w.Update(ctx, &updatedUser, []string{"Version"}, nil, db.NewOplogMsg(&userOplogMsg), db.WithVersion(&userVersion))
+			if err != nil {
+				return fmt.Errorf("set associated accounts: unable to update user version: %w", err)
+			}
+			if rowsUpdated != 1 {
+				return fmt.Errorf("set associated accounts: updated user and %d rows updated", rowsUpdated)
+			}
+
+			if len(associateIds) > 0 {
+				if err := associateUserWithAccounts(ctx, r.kms, reader, w, userId, associateIds); err != nil {
+					return fmt.Errorf("set associated accounts: unable to associate ids: %w", err)
+				}
+			}
+
+			if len(disassociateIds) > 0 {
+				if err := dissociateUserFromAccounts(ctx, r.kms, reader, w, userId, disassociateIds); err != nil {
+					return fmt.Errorf("set associated accounts: unable to disassociate ids: %w", err)
+				}
+			}
+
+			metadata := oplog.Metadata{
+				"op-type":            []string{oplog.OpType_OP_TYPE_UPDATE.String()},
+				"scope-id":           []string{user.ScopeId},
+				"scope-type":         []string{scope.Org.String()},
+				"resource-public-id": []string{userId},
+			}
+			if err := w.WriteOplogEntryWith(ctx, oplogWrapper, userTicket, metadata, []*oplog.Message{&userOplogMsg}); err != nil {
+				return fmt.Errorf("set associated accounts: unable to write oplog: %w", err)
+			}
+			// we need a new repo, that's using the same reader/writer as this TxHandler
+			txRepo := &Repository{
+				reader: reader,
+				writer: w,
+				kms:    r.kms,
+				// intentionally not setting the defaultLimit, so we'll get all
+				// the account ids without a limit
+			}
+			currentAccountIds, err = txRepo.ListAssociatedAccountIds(ctx, user.PublicId)
+			if err != nil {
+				return fmt.Errorf("set associated accounts: unable to retrieve current account ids after set: %w", err)
+			}
+			return nil
+		},
+	)
+	if err != nil {
+		return nil, fmt.Errorf("set associated accounts: %w", err)
+	}
+	return currentAccountIds, nil
+}
+
+// associateUserWithAccounts will associate the accounts (accountIds) with
+// the user (userId) within the writer's database
+func associateUserWithAccounts(ctx context.Context, repoKms *kms.Kms, reader db.Reader, writer db.Writer, userId string, accountIds []string, opt ...Option) error {
+	if repoKms == nil {
+		return fmt.Errorf("associate user with accounts: kms is nil: %w", db.ErrInvalidParameter)
+	}
+	if reader == nil {
+		return fmt.Errorf("associate user with accounts: db reader is nil: %w", db.ErrInvalidParameter)
+	}
+	if writer == nil {
+		return fmt.Errorf("associate user with accounts: db writer is nil: %w", db.ErrInvalidParameter)
+	}
+	if userId == "" {
+		return fmt.Errorf("associate user with accounts: user id is empty: %w", db.ErrInvalidParameter)
+	}
+	if len(accountIds) == 0 {
+		return fmt.Errorf("associate user with accounts: missing account id %w", db.ErrInvalidParameter)
+	}
+	authAccounts := make([]*authAccount, 0, len(accountIds))
+	for _, accountId := range accountIds {
+		acct := allocAccount()
+		acct.PublicId = accountId
+		err := reader.LookupByPublicId(context.Background(), &acct)
+		if err != nil {
+			return fmt.Errorf("associate user with accounts: unable to lookup account %s: %w", accountId, err)
+		}
+		if acct.IamUserId != "" && acct.IamUserId != userId {
+			return fmt.Errorf("associate user with accounts: %s account is associated with a user %s: %w", accountId, acct.IamUserId, db.ErrInvalidParameter)
+		}
+		authAccounts = append(authAccounts, &acct)
+	}
+
+	for _, aa := range authAccounts {
+		// wrapper could be different for each authAccount depending on it's scope
+		oplogWrapper, err := repoKms.GetWrapper(ctx, aa.GetScopeId(), kms.KeyPurposeOplog)
+		if err != nil {
+			return fmt.Errorf("associate user with accounts: unable to get oplog wrapper: %w", err)
+		}
+
+		metadata := oplog.Metadata{
+			"resource-public-id": []string{aa.PublicId},
+			"scope-id":           []string{aa.ScopeId},
+			"resource-type":      []string{"auth-account"},
+			"op-type":            []string{oplog.OpType_OP_TYPE_UPDATE.String()},
+		}
+		var updatedRows int
+		updatedAcct := aa.Clone().(*authAccount)
+		updatedAcct.IamUserId = userId
+		updatedRows, err = writer.Update(ctx, updatedAcct, []string{"IamUserId"}, nil, db.WithOplog(oplogWrapper, metadata), db.WithWhere("iam_user_id is NULL or iam_user_id = ?", userId))
+		if err != nil {
+			return fmt.Errorf("associate user with accouts: failed to associate %s account: %w", aa.PublicId, err)
+		}
+		if updatedRows == 0 {
+			return fmt.Errorf("associate user with accounts: failed to associate %s account: it is already associated with another user", aa.PublicId)
+		}
+		if updatedRows > 1 {
+			return fmt.Errorf("associate user with accounts: failed to associate %s account: would have updated too many accounts %d", aa.PublicId, updatedRows)
 		}
 	}
-
-	oplogWrapper, err := r.kms.GetWrapper(ctx, acct.GetScopeId(), kms.KeyPurposeOplog)
-	if err != nil {
-		return nil, fmt.Errorf("associate user with account: unable to get oplog wrapper: %w", err)
-	}
-
-	// validate, associated the user with the account, and then read the
-	// user back in the same tx for consistency.
-	_, err = r.writer.DoTx(
-		ctx,
-		db.StdRetryCnt,
-		db.ExpBackoff{},
-		func(txReader db.Reader, w db.Writer) error {
-			metadata := oplog.Metadata{
-				"resource-public-id": []string{accountId},
-				"scope-id":           []string{acct.ScopeId},
-				"scope-type":         []string{scope.Org.String()},
-				"resource-type":      []string{"auth-account"},
-			}
-			var updatedRows int
-			updatedAcct := acct.Clone().(*authAccount)
-			updatedAcct.IamUserId = userPublicId
-			// we are using WithWhere to make sure the account is not
-			// associated with a user (handling race conditions with concurrent
-			// transactions)
-			switch {
-			case opts.withDisassociate:
-				updatedRows, err = w.Update(ctx, updatedAcct, []string{"IamUserId"}, nil, db.WithOplog(oplogWrapper, metadata), db.WithWhere("iam_user_id = ?", acct.IamUserId))
-			default:
-				updatedRows, err = w.Update(ctx, updatedAcct, []string{"IamUserId"}, nil, db.WithOplog(oplogWrapper, metadata), db.WithWhere("iam_user_id is NULL"))
-			}
-			if err != nil {
-				return err
-			}
-			if updatedRows != 1 {
-				return fmt.Errorf("account update affected %d rows", updatedRows)
-			}
-			return nil
-		},
-	)
-	if err != nil {
-		return nil, fmt.Errorf("associate user with account: %w", err)
-	}
-	return &user, nil
+	return nil
 }
 
-// DissociateUserWithAccount will dissociate a user with its existing account.
-// An error is returned if account is associated with a different user.  No
-// options are currently supported.
-func (r *Repository) DissociateUserWithAccount(ctx context.Context, userPublicId, accountId string, opt ...Option) (*User, error) {
-	if userPublicId == "" {
-		return nil, fmt.Errorf("dissociate user with account: missing user public id %w", db.ErrInvalidParameter)
+// dissociateUserFromAccounts will dissociate a user with its existing accounts
+// (accountIds). An error is returned if account is associated with a different
+// user.  No options are currently supported.
+func dissociateUserFromAccounts(ctx context.Context, repoKms *kms.Kms, reader db.Reader, writer db.Writer, userId string, accountIds []string, opt ...Option) error {
+	if repoKms == nil {
+		return fmt.Errorf("dissociate user from accounts: kms is nil: %w", db.ErrInvalidParameter)
 	}
-	if accountId == "" {
-		return nil, fmt.Errorf("dissociate user with account: missing account id %w", db.ErrInvalidParameter)
+	if reader == nil {
+		return fmt.Errorf("dissociate user from accounts: db reader is nil: %w", db.ErrInvalidParameter)
+	}
+	if writer == nil {
+		return fmt.Errorf("dissociate user from accounts: db writer is nil: %w", db.ErrInvalidParameter)
+	}
+	if userId == "" {
+		return fmt.Errorf("dissociate user from accounts: missing user public id %w", db.ErrInvalidParameter)
+	}
+	if len(accountIds) == 0 {
+		return fmt.Errorf("dissociate user from accounts: missing account id %w", db.ErrInvalidParameter)
+	}
+	authAccounts := make([]*authAccount, 0, len(accountIds))
+	for _, accountId := range accountIds {
+		acct := allocAccount()
+		acct.PublicId = accountId
+		err := reader.LookupByPublicId(context.Background(), &acct)
+		if err != nil {
+			return fmt.Errorf("dissociate user from accounts: unable to lookup account %s: %w", accountId, err)
+		}
+		if acct.IamUserId != "" && acct.IamUserId != userId {
+			return fmt.Errorf("dissociate user from accounts: %s account is associated with a user %s: %w", accountId, acct.IamUserId, db.ErrInvalidParameter)
+		}
+		authAccounts = append(authAccounts, &acct)
 	}
 
-	user := allocUser()
-	user.PublicId = userPublicId
-	err := r.reader.LookupById(ctx, &user)
-	if err != nil {
-		return nil, fmt.Errorf("dissociate user with account: unable to lookup user %s: %w", userPublicId, err)
+	for _, aa := range authAccounts {
+		// wrapper could be different for each authAccount depending on it's scope
+		oplogWrapper, err := repoKms.GetWrapper(ctx, aa.GetScopeId(), kms.KeyPurposeOplog)
+		if err != nil {
+			return fmt.Errorf("dissociate user from accounts: unable to get oplog wrapper: %w", err)
+		}
+		metadata := oplog.Metadata{
+			"resource-public-id": []string{aa.PublicId},
+			"scope-id":           []string{aa.ScopeId},
+			"resource-type":      []string{"auth-account"},
+			"op-type":            []string{oplog.OpType_OP_TYPE_UPDATE.String()},
+		}
+		var updatedRows int
+		updatedAcct := aa.Clone().(*authAccount)
+		updatedAcct.IamUserId = userId
+		// update IamUserId to null
+		updatedRows, err = writer.Update(ctx, updatedAcct, nil, []string{"IamUserId"}, db.WithOplog(oplogWrapper, metadata), db.WithWhere("iam_user_id is NULL or iam_user_id = ?", userId))
+		if err != nil {
+			return fmt.Errorf("dissociate user from accounts: failed to disassociate %s account: %w", aa.PublicId, err)
+		}
+		if updatedRows == 0 {
+			return fmt.Errorf("dissociate user from accounts: failed to disassociate %s account: it is associated with another user", aa.PublicId)
+		}
+		if updatedRows > 1 {
+			return fmt.Errorf("dissociate user from accounts: failed to disassociate %s account: would have updated too many accounts %d", aa.PublicId, updatedRows)
+		}
 	}
-
-	acct := allocAccount()
-	acct.PublicId = accountId
-	err = r.reader.LookupByPublicId(ctx, &acct)
-	if err != nil {
-		return nil, fmt.Errorf("dissociate user with account: unable to lookup account %s: %w", accountId, err)
-	}
-	// first, let's handle the case where the account is not associated
-	// with any user, so we're done!
-	if acct.IamUserId == "" {
-		return &user, nil
-	}
-	// before proceeding with an update, is the account associated with the different user?
-	if acct.IamUserId != userPublicId {
-		return nil, fmt.Errorf("dissociate user with account: %s account is not associated with a user: %w", accountId, db.ErrInvalidParameter)
-	}
-
-	oplogWrapper, err := r.kms.GetWrapper(ctx, acct.GetScopeId(), kms.KeyPurposeOplog)
-	if err != nil {
-		return nil, fmt.Errorf("disassociate user with account: unable to get oplog wrapper: %w", err)
-	}
-
-	// validate, dissociate the user with the account and then read the user back in
-	// the same tx for consistency.
-	_, err = r.writer.DoTx(
-		ctx,
-		db.StdRetryCnt,
-		db.ExpBackoff{},
-		func(txReader db.Reader, w db.Writer) error {
-			metadata := oplog.Metadata{
-				"resource-public-id": []string{accountId},
-				"scope-id":           []string{acct.ScopeId},
-				"scope-type":         []string{scope.Org.String()},
-				"resource-type":      []string{"auth-account"},
-			}
-			updatedAcct := acct.Clone().(*authAccount)
-			updatedAcct.IamUserId = ""
-			// set the user id to null and use WithWhere to ensure that the auth
-			// account is associated with the user (handling race conditions
-			// with other concurrent transactions)
-			updatedRows, err := w.Update(ctx, updatedAcct, nil, []string{"IamUserId"}, db.WithOplog(oplogWrapper, metadata), db.WithWhere("iam_user_id = ?", userPublicId))
-			if err != nil {
-				return err
-			}
-			if updatedRows != 1 {
-				return fmt.Errorf("account update affected %d rows", updatedRows)
-			}
-			return nil
-		},
-	)
-	if err != nil {
-		return nil, fmt.Errorf("dissociate user with account: %w", err)
-	}
-	return &user, nil
+	return nil
+}
+
+// associationChanges returns two slices: accounts to associate and disassociate
+func associationChanges(ctx context.Context, reader db.Reader, userId string, accountIds []string) ([]string, []string, error) {
+	var inClauseSpots []string
+	// starts at 2 because there is already a $1 in the query
+	for i := 2; i < len(accountIds)+2; i++ {
+		inClauseSpots = append(inClauseSpots, fmt.Sprintf("$%d", i))
+	}
+	inClause := strings.Join(inClauseSpots, ",")
+	if inClause == "" {
+		inClause = "''"
+	}
+	query := fmt.Sprintf(accountChangesQuery, inClause)
+
+	var params []interface{}
+	params = append(params, userId)
+	for _, v := range accountIds {
+		params = append(params, v)
+	}
+	rows, err := reader.Query(query, params)
+	if err != nil {
+		return nil, nil, fmt.Errorf("changes: query failed: %w", err)
+	}
+	defer rows.Close()
+
+	type change struct {
+		Action    string
+		AccountId string
+	}
+	var changes []*change
+	for rows.Next() {
+		var chg change
+		if err := reader.ScanRows(rows, &chg); err != nil {
+			return nil, nil, fmt.Errorf("changes: scan row failed: %w", err)
+		}
+		changes = append(changes, &chg)
+	}
+	var associateIds, disassociateIds []string
+	for _, c := range changes {
+		if c.AccountId == "" {
+			return nil, nil, fmt.Errorf("changes: missing account id in change result")
+		}
+		switch c.Action {
+		case "associate":
+			associateIds = append(associateIds, c.AccountId)
+		case "disassociate":
+			disassociateIds = append(disassociateIds, c.AccountId)
+		default:
+			return nil, nil, fmt.Errorf("changes: unknown action %s for %s", c.Action, c.AccountId)
+		}
+
+	}
+	return associateIds, disassociateIds, nil
 }

internal/iam/repository_user_test.go

@@ -4,13 +4,16 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sort"
 	"strings"
 	"testing"
 	"time"
 
 	"github.com/hashicorp/boundary/internal/db"
 	dbassert "github.com/hashicorp/boundary/internal/db/assert"
+	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/oplog"
+	"github.com/hashicorp/boundary/sdk/strutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
@@ -97,7 +100,7 @@ func TestRepository_CreateUser(t *testing.T) {
 			assert.NotNil(u.CreateTime)
 			assert.NotNil(u.UpdateTime)
 
-			foundUser, err := repo.LookupUser(context.Background(), u.PublicId)
+			foundUser, _, err := repo.LookupUser(context.Background(), u.PublicId)
 			require.NoError(err)
 			assert.True(proto.Equal(foundUser, u))
 
@@ -115,6 +118,7 @@ func TestRepository_UpdateUser(t *testing.T) {
 	repo := TestRepo(t, conn, wrapper)
 	id := testId(t)
 	org, proj := TestScopes(t, repo)
+	authMethodId := testAuthMethod(t, conn, org.PublicId)
 	pubId := func(s string) *string { return &s }
 
 	type args struct {
@@ -298,11 +302,18 @@ func TestRepository_UpdateUser(t *testing.T) {
 			if tt.wantDup {
 				u := TestUser(t, repo, org.PublicId, tt.newUserOpts...)
 				u.Name = tt.args.name
-				_, _, err := repo.UpdateUser(context.Background(), u, 1, tt.args.fieldMaskPaths, tt.args.opt...)
+				_, _, _, err := repo.UpdateUser(context.Background(), u, 1, tt.args.fieldMaskPaths, tt.args.opt...)
 				require.NoError(err)
 			}
 
 			u := TestUser(t, repo, org.PublicId, tt.newUserOpts...)
+			acctCount := 3
+			accountIds := make([]string, 0, acctCount)
+			for i := 0; i < acctCount; i++ {
+				aa := testAccount(t, conn, org.PublicId, authMethodId, u.PublicId)
+				accountIds = append(accountIds, aa.PublicId)
+			}
+			sort.Strings(accountIds)
 
 			updateUser := allocUser()
 			updateUser.PublicId = u.PublicId
@@ -314,6 +325,7 @@ func TestRepository_UpdateUser(t *testing.T) {
 			updateUser.Description = tt.args.description
 
 			var userAfterUpdate *User
+			var acctIdsAfterUpdate []string
 			var updatedRows int
 			var err error
 			if tt.directUpdate {
@@ -324,7 +336,7 @@ func TestRepository_UpdateUser(t *testing.T) {
 					userAfterUpdate = resource.(*User)
 				}
 			} else {
-				userAfterUpdate, updatedRows, err = repo.UpdateUser(context.Background(), &updateUser, 1, tt.args.fieldMaskPaths, tt.args.opt...)
+				userAfterUpdate, acctIdsAfterUpdate, updatedRows, err = repo.UpdateUser(context.Background(), &updateUser, 1, tt.args.fieldMaskPaths, tt.args.opt...)
 			}
 			if tt.wantErr {
 				require.Error(err)
@@ -347,9 +359,14 @@ func TestRepository_UpdateUser(t *testing.T) {
 			require.NoError(err)
 			assert.Equal(tt.wantRowsUpdate, updatedRows)
 			assert.NotEqual(u.UpdateTime, userAfterUpdate.UpdateTime)
-			foundUser, err := repo.LookupUser(context.Background(), u.PublicId)
+			sort.Strings(acctIdsAfterUpdate)
+			assert.Equal(accountIds, acctIdsAfterUpdate)
+
+			foundUser, foundAccountIds, err := repo.LookupUser(context.Background(), u.PublicId)
 			require.NoError(err)
 			assert.True(proto.Equal(userAfterUpdate, foundUser))
+			sort.Strings(foundAccountIds)
+			assert.Equal(accountIds, foundAccountIds)
 
 			dbassert := dbassert.New(t, rw)
 			if tt.args.name == "" {
@@ -436,7 +453,7 @@ func TestRepository_DeleteUser(t *testing.T) {
 			}
 			require.NoError(err)
 			assert.Equal(tt.wantRowsDeleted, deletedRows)
-			foundUser, err := repo.LookupUser(context.Background(), tt.args.user.PublicId)
+			foundUser, _, err := repo.LookupUser(context.Background(), tt.args.user.PublicId)
 			require.NoError(err)
 			assert.Nil(foundUser)
 
@@ -664,18 +681,19 @@ func TestRepository_LookupUserWithLogin(t *testing.T) {
 	}
 }
 
-func TestRepository_AssociateUserWithAccount(t *testing.T) {
+func TestRepository_associateUserWithAccounts(t *testing.T) {
 	t.Parallel()
 	conn, _ := db.TestSetup(t, "postgres")
 	rw := db.New(conn)
 	wrapper := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrapper)
 	repo := TestRepo(t, conn, wrapper)
 	org, _ := TestScopes(t, repo)
 	authMethodId := testAuthMethod(t, conn, org.PublicId)
 
 	type Ids struct {
-		user string
-		acct string
+		user  string
+		accts []string
 	}
 	type args struct {
 		Ids Ids
@@ -696,7 +714,8 @@ func TestRepository_AssociateUserWithAccount(t *testing.T) {
 				Ids: func() Ids {
 					u := TestUser(t, repo, org.PublicId)
 					a := testAccount(t, conn, org.PublicId, authMethodId, "")
-					return Ids{user: u.PublicId, acct: a.PublicId}
+					a2 := testAccount(t, conn, org.PublicId, authMethodId, "")
+					return Ids{user: u.PublicId, accts: []string{a.PublicId, a2.PublicId}}
 				}(),
 			},
 		},
@@ -717,7 +736,7 @@ func TestRepository_AssociateUserWithAccount(t *testing.T) {
 			args: args{
 				Ids: func() Ids {
 					id := testId(t)
-					return Ids{acct: id}
+					return Ids{accts: []string{id}}
 				}(),
 			},
 			wantErr:   true,
@@ -729,7 +748,7 @@ func TestRepository_AssociateUserWithAccount(t *testing.T) {
 				Ids: func() Ids {
 					u := TestUser(t, repo, org.PublicId)
 					a := testAccount(t, conn, org.PublicId, authMethodId, u.PublicId)
-					return Ids{user: u.PublicId, acct: a.PublicId}
+					return Ids{user: u.PublicId, accts: []string{a.PublicId}}
 				}(),
 				opt: []Option{WithDisassociate(true)},
 			},
@@ -741,7 +760,7 @@ func TestRepository_AssociateUserWithAccount(t *testing.T) {
 					u := TestUser(t, repo, org.PublicId)
 					diffUser := TestUser(t, repo, org.PublicId)
 					a := testAccount(t, conn, org.PublicId, authMethodId, diffUser.PublicId)
-					return Ids{user: u.PublicId, acct: a.PublicId}
+					return Ids{user: u.PublicId, accts: []string{a.PublicId}}
 				}(),
 			},
 			wantErr:   true,
@@ -754,7 +773,7 @@ func TestRepository_AssociateUserWithAccount(t *testing.T) {
 					u := TestUser(t, repo, org.PublicId)
 					diffUser := TestUser(t, repo, org.PublicId)
 					a := testAccount(t, conn, org.PublicId, authMethodId, diffUser.PublicId)
-					return Ids{user: u.PublicId, acct: a.PublicId}
+					return Ids{user: u.PublicId, accts: []string{a.PublicId}}
 				}(),
 			},
 			wantErr:   true,
@@ -766,60 +785,72 @@ func TestRepository_AssociateUserWithAccount(t *testing.T) {
 				Ids: func() Ids {
 					u := TestUser(t, repo, org.PublicId)
 					id := testId(t)
-					return Ids{user: u.PublicId, acct: id}
+					return Ids{user: u.PublicId, accts: []string{id}}
 				}(),
 			},
 			wantErr:   true,
 			wantErrIs: db.ErrRecordNotFound,
 		},
 		{
-			name: "bad-user-id",
+			name: "bad-user-id-not-associated-account",
 			args: args{
 				Ids: func() Ids {
 					id := testId(t)
 					a := testAccount(t, conn, org.PublicId, authMethodId, "")
-					return Ids{user: id, acct: a.PublicId}
+					return Ids{user: id, accts: []string{a.PublicId}}
 				}(),
 			},
-			wantErr:   true,
-			wantErrIs: db.ErrRecordNotFound,
+			wantErr: false,
+		},
+		{
+			name: "bad-user-id",
+			args: args{
+				Ids: func() Ids {
+					id := testId(t)
+					testUser := TestUser(t, repo, org.PublicId)
+					a := testAccount(t, conn, org.PublicId, authMethodId, testUser.PublicId)
+					return Ids{user: id, accts: []string{a.PublicId}}
+				}(),
+			},
+			wantErr: true,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			assert, require := assert.New(t), require.New(t)
-			u, err := repo.AssociateUserWithAccount(context.Background(), tt.args.Ids.user, tt.args.Ids.acct, tt.args.opt...)
+			err := associateUserWithAccounts(context.Background(), kms, rw, rw, tt.args.Ids.user, tt.args.Ids.accts, tt.args.opt...)
 			if tt.wantErr {
 				require.Error(err)
-				assert.Empty(u)
 				if tt.wantErrIs != nil {
 					assert.Truef(errors.Is(err, tt.wantErrIs), "unexpected error %s", err.Error())
 				}
 				return
 			}
 			require.NoError(err)
-			assert.Equal(tt.args.Ids.user, u.PublicId)
-			acct := allocAccount()
-			acct.PublicId = tt.args.Ids.acct
-			err = rw.LookupByPublicId(context.Background(), &acct)
-			require.NoError(err)
-			assert.Equal(tt.args.Ids.user, acct.IamUserId)
+			for _, id := range tt.args.Ids.accts {
+				acct := allocAccount()
+				acct.PublicId = id
+				err = rw.LookupByPublicId(context.Background(), &acct)
+				require.NoError(err)
+				assert.Equal(tt.args.Ids.user, acct.IamUserId)
+			}
 		})
 	}
 }
 
-func TestRepository_DissociateUserWithAccount(t *testing.T) {
+func TestRepository_dissociateUserWithAccount(t *testing.T) {
 	t.Parallel()
 	conn, _ := db.TestSetup(t, "postgres")
 	rw := db.New(conn)
 	wrapper := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrapper)
 	repo := TestRepo(t, conn, wrapper)
 	org, _ := TestScopes(t, repo)
 	authMethodId := testAuthMethod(t, conn, org.PublicId)
 
 	type Ids struct {
-		user string
-		acct string
+		user  string
+		accts []string
 	}
 	type args struct {
 		Ids Ids
@@ -839,7 +870,8 @@ func TestRepository_DissociateUserWithAccount(t *testing.T) {
 				Ids: func() Ids {
 					u := TestUser(t, repo, org.PublicId)
 					a := testAccount(t, conn, org.PublicId, authMethodId, u.PublicId)
-					return Ids{user: u.PublicId, acct: a.PublicId}
+					a2 := testAccount(t, conn, org.PublicId, authMethodId, "")
+					return Ids{user: u.PublicId, accts: []string{a.PublicId, a2.PublicId}}
 				}(),
 			},
 		},
@@ -860,7 +892,7 @@ func TestRepository_DissociateUserWithAccount(t *testing.T) {
 			args: args{
 				Ids: func() Ids {
 					id := testId(t)
-					return Ids{acct: id}
+					return Ids{accts: []string{id}}
 				}(),
 			},
 			wantErr:   true,
@@ -872,7 +904,7 @@ func TestRepository_DissociateUserWithAccount(t *testing.T) {
 				Ids: func() Ids {
 					u := TestUser(t, repo, org.PublicId)
 					a := testAccount(t, conn, org.PublicId, authMethodId, "")
-					return Ids{user: u.PublicId, acct: a.PublicId}
+					return Ids{user: u.PublicId, accts: []string{a.PublicId}}
 				}(),
 			},
 		},
@@ -883,7 +915,7 @@ func TestRepository_DissociateUserWithAccount(t *testing.T) {
 					u := TestUser(t, repo, org.PublicId)
 					diffUser := TestUser(t, repo, org.PublicId)
 					a := testAccount(t, conn, org.PublicId, authMethodId, diffUser.PublicId)
-					return Ids{user: u.PublicId, acct: a.PublicId}
+					return Ids{user: u.PublicId, accts: []string{a.PublicId}}
 				}(),
 			},
 			wantErr:   true,
@@ -895,45 +927,517 @@ func TestRepository_DissociateUserWithAccount(t *testing.T) {
 				Ids: func() Ids {
 					u := TestUser(t, repo, org.PublicId)
 					id := testId(t)
-					return Ids{user: u.PublicId, acct: id}
+					return Ids{user: u.PublicId, accts: []string{id}}
 				}(),
 			},
 			wantErr:   true,
 			wantErrIs: db.ErrRecordNotFound,
 		},
 		{
-			name: "bad-user-id",
+			name: "bad-user-id-not-associated-account",
 			args: args{
 				Ids: func() Ids {
 					id := testId(t)
 					a := testAccount(t, conn, org.PublicId, authMethodId, "")
-					return Ids{user: id, acct: a.PublicId}
+					return Ids{user: id, accts: []string{a.PublicId}}
 				}(),
 			},
-			wantErr:   true,
-			wantErrIs: db.ErrRecordNotFound,
+			wantErr: false,
+		},
+		{
+			name: "bad-user-id",
+			args: args{
+				Ids: func() Ids {
+					id := testId(t)
+					testUser := TestUser(t, repo, org.PublicId)
+					a := testAccount(t, conn, org.PublicId, authMethodId, testUser.PublicId)
+					return Ids{user: id, accts: []string{a.PublicId}}
+				}(),
+			},
+			wantErr: true,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			assert, require := assert.New(t), require.New(t)
 			dbassert := dbassert.New(t, rw)
-			u, err := repo.DissociateUserWithAccount(context.Background(), tt.args.Ids.user, tt.args.Ids.acct, tt.args.opt...)
+			err := dissociateUserFromAccounts(context.Background(), kms, rw, rw, tt.args.Ids.user, tt.args.Ids.accts, tt.args.opt...)
 			if tt.wantErr {
 				require.Error(err)
-				assert.Empty(u)
 				if tt.wantErrIs != nil {
 					assert.Truef(errors.Is(err, tt.wantErrIs), "unexpected error %s", err.Error())
 				}
 				return
 			}
 			require.NoError(err)
-			assert.Equal(tt.args.Ids.user, u.PublicId)
-
-			acct := allocAccount()
-			acct.PublicId = tt.args.Ids.acct
-			dbassert.IsNull(&acct, "IamUserId")
 
+			for _, id := range tt.args.Ids.accts {
+				acct := allocAccount()
+				acct.PublicId = id
+				dbassert.IsNull(&acct, "IamUserId")
+			}
+		})
+	}
+}
+
+func TestRepository_AssociateAccounts(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	repo := TestRepo(t, conn, wrapper)
+	org, _ := TestScopes(t, repo)
+	authMethodId := testAuthMethod(t, conn, org.PublicId)
+	user := TestUser(t, repo, org.PublicId)
+
+	createAccountsFn := func() []string {
+		require.NoError(t, conn.Where("iam_user_id = ?", user.PublicId).Delete(allocAccount()).Error)
+		results := []string{}
+		for i := 0; i < 5; i++ {
+			a := testAccount(t, conn, org.PublicId, authMethodId, "")
+			results = append(results, a.PublicId)
+		}
+		return results
+	}
+	type args struct {
+		accountIdsFn        func() []string
+		userId              string
+		userVersionOverride *uint32
+		opt                 []Option
+	}
+	tests := []struct {
+		name      string
+		args      args
+		wantErr   bool
+		wantErrIs error
+	}{
+		{
+			name: "valid",
+			args: args{
+				userId:       user.PublicId,
+				accountIdsFn: createAccountsFn,
+			},
+			wantErr: false,
+		},
+		{
+			name: "already-associated",
+			args: args{
+				userId: user.PublicId,
+				accountIdsFn: func() []string {
+					ids := createAccountsFn()
+					a := testAccount(t, conn, org.PublicId, authMethodId, user.PublicId)
+					ids = append(ids, a.PublicId)
+					return ids
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "associated-with-diff-user",
+			args: args{
+				userId: user.PublicId,
+				accountIdsFn: func() []string {
+					ids := createAccountsFn()
+					u := TestUser(t, repo, org.PublicId)
+					a := testAccount(t, conn, org.PublicId, authMethodId, u.PublicId)
+					ids = append(ids, a.PublicId)
+					return ids
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "bad-version",
+			args: args{
+				userVersionOverride: func() *uint32 {
+					i := uint32(22)
+					return &i
+				}(),
+				userId:       user.PublicId,
+				accountIdsFn: createAccountsFn,
+			},
+			wantErr: true,
+		},
+		{
+			name: "zero-version",
+			args: args{
+				userVersionOverride: func() *uint32 {
+					i := uint32(0)
+					return &i
+				}(),
+				userId:       user.PublicId,
+				accountIdsFn: createAccountsFn,
+			},
+			wantErr: true,
+		},
+		{
+			name: "no-accounts",
+			args: args{
+				userId:       user.PublicId,
+				accountIdsFn: func() []string { return nil },
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert, require := assert.New(t), require.New(t)
+			accountIds := tt.args.accountIdsFn()
+			sort.Strings(accountIds)
+
+			origUser, _, err := repo.LookupUser(context.Background(), user.PublicId)
+			require.NoError(err)
+
+			version := origUser.Version
+			if tt.args.userVersionOverride != nil {
+				version = *tt.args.userVersionOverride
+			}
+
+			got, err := repo.AssociateAccounts(context.Background(), tt.args.userId, version, accountIds, tt.args.opt...)
+			if tt.wantErr {
+				require.Error(err)
+				if tt.wantErrIs != nil {
+					assert.Truef(errors.Is(err, tt.wantErrIs), "unexpected error %s", err.Error())
+				}
+				return
+			}
+			require.NoError(err)
+			err = db.TestVerifyOplog(t, rw, tt.args.userId, db.WithOperation(oplog.OpType_OP_TYPE_UPDATE), db.WithCreateNotBefore(10*time.Second))
+			assert.NoError(err)
+			for _, id := range got {
+				err = db.TestVerifyOplog(t, rw, id, db.WithOperation(oplog.OpType_OP_TYPE_UPDATE), db.WithCreateNotBefore(10*time.Second))
+				assert.NoError(err)
+			}
+
+			sort.Strings(got)
+			assert.Equal(accountIds, got)
+
+			foundIds, err := repo.ListAssociatedAccountIds(context.Background(), tt.args.userId)
+			require.NoError(err)
+			sort.Strings(foundIds)
+			assert.Equal(accountIds, foundIds)
+
+			u, _, err := repo.LookupUser(context.Background(), tt.args.userId)
+			require.NoError(err)
+			assert.Equal(version+1, u.Version)
+		})
+	}
+}
+
+func TestRepository_DisassociateAccounts(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	repo := TestRepo(t, conn, wrapper)
+	org, _ := TestScopes(t, repo)
+	authMethodId := testAuthMethod(t, conn, org.PublicId)
+	user := TestUser(t, repo, org.PublicId)
+
+	createAccountsFn := func() []string {
+		require.NoError(t, conn.Where("iam_user_id = ?", user.PublicId).Delete(allocAccount()).Error)
+		results := []string{}
+		for i := 0; i < 1; i++ {
+			a := testAccount(t, conn, org.PublicId, authMethodId, user.PublicId)
+			results = append(results, a.PublicId)
+		}
+		return results
+	}
+	type args struct {
+		accountIdsFn        func() []string
+		userId              string
+		userVersionOverride *uint32
+		opt                 []Option
+	}
+	tests := []struct {
+		name      string
+		args      args
+		wantErr   bool
+		wantErrIs error
+	}{
+		{
+			name: "valid",
+			args: args{
+				userId:       user.PublicId,
+				accountIdsFn: createAccountsFn,
+			},
+			wantErr: false,
+		},
+		{
+			name: "associated-with-diff-user",
+			args: args{
+				userId: user.PublicId,
+				accountIdsFn: func() []string {
+					ids := createAccountsFn()
+					u := TestUser(t, repo, org.PublicId)
+					a := testAccount(t, conn, org.PublicId, authMethodId, u.PublicId)
+					ids = append(ids, a.PublicId)
+					return ids
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "bad-version",
+			args: args{
+				userVersionOverride: func() *uint32 {
+					i := uint32(22)
+					return &i
+				}(),
+				userId:       user.PublicId,
+				accountIdsFn: createAccountsFn,
+			},
+			wantErr: true,
+		},
+		{
+			name: "zero-version",
+			args: args{
+				userVersionOverride: func() *uint32 {
+					i := uint32(0)
+					return &i
+				}(),
+				userId:       user.PublicId,
+				accountIdsFn: createAccountsFn,
+			},
+			wantErr: true,
+		},
+		{
+			name: "no-accounts",
+			args: args{
+				userId:       user.PublicId,
+				accountIdsFn: func() []string { return nil },
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert, require := assert.New(t), require.New(t)
+			accountIds := tt.args.accountIdsFn()
+
+			origUser, _, err := repo.LookupUser(context.Background(), user.PublicId)
+			require.NoError(err)
+
+			version := origUser.Version
+			if tt.args.userVersionOverride != nil {
+				version = *tt.args.userVersionOverride
+			}
+
+			got, err := repo.DisassociateAccounts(context.Background(), tt.args.userId, version, accountIds, tt.args.opt...)
+			if tt.wantErr {
+				require.Error(err)
+				if tt.wantErrIs != nil {
+					assert.Truef(errors.Is(err, tt.wantErrIs), "unexpected error %s", err.Error())
+				}
+				return
+			}
+			require.NoError(err)
+			err = db.TestVerifyOplog(t, rw, tt.args.userId, db.WithOperation(oplog.OpType_OP_TYPE_UPDATE), db.WithCreateNotBefore(10*time.Second))
+			assert.NoError(err)
+			for _, id := range got {
+				err = db.TestVerifyOplog(t, rw, id, db.WithOperation(oplog.OpType_OP_TYPE_UPDATE), db.WithCreateNotBefore(10*time.Second))
+				assert.NoError(err)
+			}
+			foundIds, err := repo.ListAssociatedAccountIds(context.Background(), tt.args.userId)
+			require.NoError(err)
+			for _, id := range accountIds {
+				assert.True(!strutil.StrListContains(foundIds, id))
+			}
+
+			u, _, err := repo.LookupUser(context.Background(), tt.args.userId)
+			require.NoError(err)
+			assert.Equal(version+1, u.Version)
+		})
+	}
+}
+
+func TestRepository_SetAssociatedAccounts(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	repo := TestRepo(t, conn, wrapper)
+	org, _ := TestScopes(t, repo)
+	authMethodId := testAuthMethod(t, conn, org.PublicId)
+	user := TestUser(t, repo, org.PublicId)
+
+	createAccountsFn := func() []string {
+		require.NoError(t, conn.Where("iam_user_id = ?", user.PublicId).Delete(allocAccount()).Error)
+		results := []string{}
+		for i := 0; i < 5; i++ {
+			a := testAccount(t, conn, org.PublicId, authMethodId, "")
+			results = append(results, a.PublicId)
+		}
+		return results
+	}
+	type args struct {
+		accountIdsFn        func() ([]string, []string)
+		userId              string
+		userVersionOverride *uint32
+		opt                 []Option
+	}
+	tests := []struct {
+		name      string
+		args      args
+		wantErr   bool
+		wantErrIs error
+	}{
+		{
+			name: "valid",
+			args: args{
+				userId: user.PublicId,
+				accountIdsFn: func() ([]string, []string) {
+					ids := createAccountsFn()
+					return ids, ids
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "one-already-associated",
+			args: args{
+				userId: user.PublicId,
+				accountIdsFn: func() ([]string, []string) {
+					ids := createAccountsFn()
+					changes := append([]string{}, ids...)
+					a := testAccount(t, conn, org.PublicId, authMethodId, user.PublicId)
+					ids = append(ids, a.PublicId)
+					return ids, changes
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "no-change",
+			args: args{
+				userId: user.PublicId,
+				accountIdsFn: func() ([]string, []string) {
+					ids := []string{}
+					for i := 0; i < 10; i++ {
+						a := testAccount(t, conn, org.PublicId, authMethodId, user.PublicId)
+						ids = append(ids, a.PublicId)
+					}
+					return ids, nil
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "remove-all",
+			args: args{
+				userId: user.PublicId,
+				accountIdsFn: func() ([]string, []string) {
+					ids := []string{}
+					for i := 0; i < 10; i++ {
+						a := testAccount(t, conn, org.PublicId, authMethodId, user.PublicId)
+						ids = append(ids, a.PublicId)
+					}
+					return nil, ids
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "associated-with-diff-user",
+			args: args{
+				userId: user.PublicId,
+				accountIdsFn: func() ([]string, []string) {
+					ids := createAccountsFn()
+					u := TestUser(t, repo, org.PublicId)
+					a := testAccount(t, conn, org.PublicId, authMethodId, u.PublicId)
+					ids = append(ids, a.PublicId)
+					return ids, ids
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "bad-version",
+			args: args{
+				userVersionOverride: func() *uint32 {
+					i := uint32(22)
+					return &i
+				}(),
+				userId: user.PublicId,
+				accountIdsFn: func() ([]string, []string) {
+					ids := createAccountsFn()
+					return ids, ids
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "zero-version",
+			args: args{
+				userVersionOverride: func() *uint32 {
+					i := uint32(0)
+					return &i
+				}(),
+				userId: user.PublicId,
+				accountIdsFn: func() ([]string, []string) {
+					ids := createAccountsFn()
+					return ids, ids
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "no-accounts-no-changes",
+			args: args{
+				userId:       user.PublicId,
+				accountIdsFn: func() ([]string, []string) { return nil, nil },
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert, require := assert.New(t), require.New(t)
+			accountIds, changes := tt.args.accountIdsFn()
+			sort.Strings(accountIds)
+
+			origUser, _, err := repo.LookupUser(context.Background(), user.PublicId)
+			require.NoError(err)
+
+			version := origUser.Version
+			if tt.args.userVersionOverride != nil {
+				version = *tt.args.userVersionOverride
+			}
+
+			got, err := repo.SetAssociatedAccounts(context.Background(), tt.args.userId, version, accountIds, tt.args.opt...)
+			if tt.wantErr {
+				require.Error(err)
+				if tt.wantErrIs != nil {
+					assert.Truef(errors.Is(err, tt.wantErrIs), "unexpected error %s", err.Error())
+				}
+				return
+			}
+			require.NoError(err)
+			if len(changes) != 0 {
+				err = db.TestVerifyOplog(t, rw, tt.args.userId, db.WithOperation(oplog.OpType_OP_TYPE_UPDATE), db.WithCreateNotBefore(10*time.Second))
+				assert.NoError(err)
+				for _, id := range changes {
+					err = db.TestVerifyOplog(t, rw, id, db.WithOperation(oplog.OpType_OP_TYPE_UPDATE), db.WithCreateNotBefore(10*time.Second))
+					assert.NoErrorf(err, "%s missing oplog entry", id)
+				}
+			}
+
+			sort.Strings(got)
+			assert.Equal(accountIds, got)
+
+			foundIds, err := repo.ListAssociatedAccountIds(context.Background(), tt.args.userId)
+			require.NoError(err)
+			sort.Strings(foundIds)
+			assert.Equal(accountIds, foundIds)
+
+			u, _, err := repo.LookupUser(context.Background(), tt.args.userId)
+			require.NoError(err)
+			switch tt.name {
+			case "no-accounts-no-changes":
+				assert.Equal(version, u.Version)
+			default:
+				assert.Equal(version+1, u.Version)
+			}
 		})
 	}
 }

internal/iam/user_test.go

@@ -230,7 +230,7 @@ func Test_UserUpdate(t *testing.T) {
 			require.NoError(err)
 			assert.Equal(tt.wantRowsUpdate, updatedRows)
 			assert.NotEqual(u.UpdateTime, updateUser.UpdateTime)
-			foundUser, err := repo.LookupUser(context.Background(), u.PublicId)
+			foundUser, _, err := repo.LookupUser(context.Background(), u.PublicId)
 			require.NoError(err)
 			assert.True(proto.Equal(updateUser, foundUser))
 		})
@@ -291,7 +291,7 @@ func Test_UserDelete(t *testing.T) {
 			}
 			require.NoError(err)
 			assert.Equal(tt.wantRowsDeleted, deletedRows)
-			foundUser, err := repo.LookupUser(context.Background(), tt.user.GetPublicId())
+			foundUser, _, err := repo.LookupUser(context.Background(), tt.user.GetPublicId())
 			assert.NoError(err)
 			assert.Nil(foundUser)
 		})

internal/servers/controller/handlers/users/user_service.go

@@ -138,7 +138,7 @@ func (s Service) getFromRepo(ctx context.Context, id string) (*pb.User, error) {
 	if err != nil {
 		return nil, err
 	}
-	u, err := repo.LookupUser(ctx, id)
+	u, _, err := repo.LookupUser(ctx, id)
 	if err != nil {
 		if errors.Is(err, db.ErrRecordNotFound) {
 			return nil, handlers.NotFoundErrorf("User %q doesn't exist.", id)
@@ -199,7 +199,7 @@ func (s Service) updateInRepo(ctx context.Context, orgId, id string, mask []stri
 	if err != nil {
 		return nil, err
 	}
-	out, rowsUpdated, err := repo.UpdateUser(ctx, u, version, dbMask)
+	out, _, rowsUpdated, err := repo.UpdateUser(ctx, u, version, dbMask)
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "Unable to update user: %v.", err)
 	}
@@ -263,7 +263,7 @@ func (s Service) authResult(ctx context.Context, id string, a action.Type) auth.
 			return res
 		}
 	default:
-		u, err := repo.LookupUser(ctx, id)
+		u, _, err := repo.LookupUser(ctx, id)
 		if err != nil {
 			res.Error = err
 			return res

internal/servers/controller/handlers/users/user_service_test.go

@@ -356,7 +356,7 @@ func TestUpdate(t *testing.T) {
 		repo, err := repoFn()
 		require.NoError(t, err, "Couldn't get a new repo")
 		version++ // From the test case that resulted in calling this
-		u, _, err = repo.UpdateUser(context.Background(), u, version, []string{"Name", "Description"})
+		u, _, _, err = repo.UpdateUser(context.Background(), u, version, []string{"Name", "Description"})
 		require.NoError(t, err, "Failed to reset the user")
 		version++
 	}