diff --git a/Makefile b/Makefile
index aac8bda01..306fb3f8b 100644
--- a/Makefile
+++ b/Makefile
@@ -67,6 +67,7 @@ protobuild:
 	@protoc --proto_path=internal/proto/local --proto_path=internal/proto/third_party --swagger_out=logtostderr=true,disable_default_errors=true,include_package_in_tags=true,fqn_for_swagger_name=true,allow_merge,merge_file_name=controller:internal/gen/. internal/proto/local/controller/api/services/v1/*.proto
 	@protoc-go-inject-tag -input=./internal/oplog/store/oplog.pb.go
 	@protoc-go-inject-tag -input=./internal/oplog/oplog_test/oplog_test.pb.go
+	@protoc-go-inject-tag -input=./internal/iam/store/group_member.pb.go
 	@protoc-go-inject-tag -input=./internal/iam/store/group.pb.go
 	@protoc-go-inject-tag -input=./internal/iam/store/role.pb.go
 	@protoc-go-inject-tag -input=./internal/iam/store/principal_role.pb.go
diff --git a/internal/db/migrations/postgres.gen.go b/internal/db/migrations/postgres.gen.go
index e38c47265..86a6c7b6b 100644
--- a/internal/db/migrations/postgres.gen.go
+++ b/internal/db/migrations/postgres.gen.go
@@ -266,7 +266,7 @@ values
   ('iam_scope', 1),
   ('iam_user', 1),
   ('iam_group', 1),
-  ('iam_group_member_user', 1),
+  ('iam_group_member', 1),
   ('iam_role', 1),
   ('iam_role_grant', 1),
   ('iam_group_role', 1),
@@ -446,6 +446,8 @@ drop table iam_role cascade;
 drop view iam_principal_role cascade;
 drop table iam_group_role cascade;
 drop table iam_user_role cascade;
+drop table iam_group_member_user cascade;
+drop view iam_group_member cascade;
 drop table iam_role_grant cascade;
 
 drop function iam_sub_names cascade;
@@ -453,6 +455,10 @@ drop function iam_immutable_scope_type_func cascade;
 drop function iam_sub_scopes_func cascade;
 drop function iam_immutable_role cascade;
 drop function iam_user_role_scope_check cascade;
+drop function iam_group_role_scope_check cascade;
+drop function iam_group_member_scope_check cascade;
+drop function iam_immutable_group_member cascade;
+drop function get_scoped_member_id cascade;
 drop function grant_scope_id_valid cascade;
 drop function immutable_scope_id_func cascade;
 drop function disallow_global_scope_deletion cascade;
@@ -867,7 +873,8 @@ before
 insert on iam_role_grant
   for each row execute procedure default_create_time();
 
-create trigger immutable_scope_id
+-- prefix a_ to make this trigger run before ensure_grant_scope_id_valid
+create trigger a_immutable_scope_id
 before
 update on iam_role
   for each row execute procedure iam_immutable_scope_id_func();
@@ -909,10 +916,19 @@ create table iam_group (
     scope_id wt_scope_id not null references iam_scope(public_id) on delete cascade on update cascade,
     unique(name, scope_id),
     disabled boolean not null default false,
+    -- version allows optimistic locking of the group when modifying the group
+    -- itself and when modifying dependent items like group members. 
+    version bigint not null default 1,
+
     -- add unique index so a composite fk can be declared.
     unique(scope_id, public_id)
   );
   
+create trigger 
+  update_version_column
+after update on iam_group
+  for each row execute procedure update_version_column();
+
 create trigger 
   update_time_column 
 before update on iam_group
@@ -1060,6 +1076,74 @@ before
 insert on iam_group_role
   for each row execute procedure default_create_time();
 
+-- iam_group_member_user is an association table that represents groups with
+-- associated users.
+create table iam_group_member_user (
+  create_time wt_timestamp,
+  group_id wt_public_id references iam_group(public_id) on delete cascade on update cascade,
+  member_id wt_public_id references iam_user(public_id) on delete cascade on update cascade,
+  primary key (group_id, member_id)
+);
+
+-- iam_immutable_group_member() ensures that group members are immutable. 
+create or replace function
+  iam_immutable_group_member()
+  returns trigger
+as $$
+begin
+    raise exception 'group members are immutable';
+end;
+$$ language plpgsql;
+
+create trigger 
+  default_create_time_column
+before
+insert on iam_group_member_user
+  for each row execute procedure default_create_time();
+
+create trigger 
+  immutable_create_time
+before
+update on iam_group_member_user
+  for each row execute procedure immutable_create_time_func();
+
+create trigger iam_immutable_group_member
+before
+update on iam_group_member_user
+  for each row execute procedure iam_immutable_group_member();
+
+-- get_scoped_principal_id is used by the iam_group_member view as a convient
+-- way to create <scope_id>:<member_id> to reference members from
+-- other scopes than the group's scope. 
+create or replace function get_scoped_member_id(group_scope text, member_scope text, member_id text) returns text 
+as $$
+begin
+	if group_scope = member_scope then
+		return member_id;
+	end if;
+	return member_scope || ':' || member_id;
+end;
+$$ language plpgsql;
+
+-- iam_group_member provides a consolidated view of group members.
+create view iam_group_member as
+select
+  gm.create_time,
+  gm.group_id,
+  gm.member_id,
+  u.scope_id as member_scope_id,
+  g.scope_id as group_scope_id,
+  get_scoped_member_id(g.scope_id, u.scope_id, gm.member_id) as scoped_member_id,
+  'user' as type
+from
+  iam_group_member_user gm,
+  iam_user u,
+  iam_group g
+where
+  gm.member_id = u.public_id and
+  gm.group_id = g.public_id;
+  
+
 commit;
 
 `),
diff --git a/internal/db/migrations/postgres/02_oplog.up.sql b/internal/db/migrations/postgres/02_oplog.up.sql
index ba84a1241..fed8b1314 100644
--- a/internal/db/migrations/postgres/02_oplog.up.sql
+++ b/internal/db/migrations/postgres/02_oplog.up.sql
@@ -91,7 +91,7 @@ values
   ('iam_scope', 1),
   ('iam_user', 1),
   ('iam_group', 1),
-  ('iam_group_member_user', 1),
+  ('iam_group_member', 1),
   ('iam_role', 1),
   ('iam_role_grant', 1),
   ('iam_group_role', 1),
diff --git a/internal/db/migrations/postgres/06_iam.down.sql b/internal/db/migrations/postgres/06_iam.down.sql
index e4c615ade..37cb91837 100644
--- a/internal/db/migrations/postgres/06_iam.down.sql
+++ b/internal/db/migrations/postgres/06_iam.down.sql
@@ -11,6 +11,8 @@ drop table iam_role cascade;
 drop view iam_principal_role cascade;
 drop table iam_group_role cascade;
 drop table iam_user_role cascade;
+drop table iam_group_member_user cascade;
+drop view iam_group_member cascade;
 drop table iam_role_grant cascade;
 
 drop function iam_sub_names cascade;
@@ -18,6 +20,10 @@ drop function iam_immutable_scope_type_func cascade;
 drop function iam_sub_scopes_func cascade;
 drop function iam_immutable_role cascade;
 drop function iam_user_role_scope_check cascade;
+drop function iam_group_role_scope_check cascade;
+drop function iam_group_member_scope_check cascade;
+drop function iam_immutable_group_member cascade;
+drop function get_scoped_member_id cascade;
 drop function grant_scope_id_valid cascade;
 drop function immutable_scope_id_func cascade;
 drop function disallow_global_scope_deletion cascade;
diff --git a/internal/db/migrations/postgres/06_iam.up.sql b/internal/db/migrations/postgres/06_iam.up.sql
index 4b40abc43..19c352746 100644
--- a/internal/db/migrations/postgres/06_iam.up.sql
+++ b/internal/db/migrations/postgres/06_iam.up.sql
@@ -441,10 +441,19 @@ create table iam_group (
     scope_id wt_scope_id not null references iam_scope(public_id) on delete cascade on update cascade,
     unique(name, scope_id),
     disabled boolean not null default false,
+    -- version allows optimistic locking of the group when modifying the group
+    -- itself and when modifying dependent items like group members. 
+    version bigint not null default 1,
+
     -- add unique index so a composite fk can be declared.
     unique(scope_id, public_id)
   );
   
+create trigger 
+  update_version_column
+after update on iam_group
+  for each row execute procedure update_version_column();
+
 create trigger 
   update_time_column 
 before update on iam_group
@@ -592,4 +601,72 @@ before
 insert on iam_group_role
   for each row execute procedure default_create_time();
 
+-- iam_group_member_user is an association table that represents groups with
+-- associated users.
+create table iam_group_member_user (
+  create_time wt_timestamp,
+  group_id wt_public_id references iam_group(public_id) on delete cascade on update cascade,
+  member_id wt_public_id references iam_user(public_id) on delete cascade on update cascade,
+  primary key (group_id, member_id)
+);
+
+-- iam_immutable_group_member() ensures that group members are immutable. 
+create or replace function
+  iam_immutable_group_member()
+  returns trigger
+as $$
+begin
+    raise exception 'group members are immutable';
+end;
+$$ language plpgsql;
+
+create trigger 
+  default_create_time_column
+before
+insert on iam_group_member_user
+  for each row execute procedure default_create_time();
+
+create trigger 
+  immutable_create_time
+before
+update on iam_group_member_user
+  for each row execute procedure immutable_create_time_func();
+
+create trigger iam_immutable_group_member
+before
+update on iam_group_member_user
+  for each row execute procedure iam_immutable_group_member();
+
+-- get_scoped_principal_id is used by the iam_group_member view as a convient
+-- way to create <scope_id>:<member_id> to reference members from
+-- other scopes than the group's scope. 
+create or replace function get_scoped_member_id(group_scope text, member_scope text, member_id text) returns text 
+as $$
+begin
+	if group_scope = member_scope then
+		return member_id;
+	end if;
+	return member_scope || ':' || member_id;
+end;
+$$ language plpgsql;
+
+-- iam_group_member provides a consolidated view of group members.
+create view iam_group_member as
+select
+  gm.create_time,
+  gm.group_id,
+  gm.member_id,
+  u.scope_id as member_scope_id,
+  g.scope_id as group_scope_id,
+  get_scoped_member_id(g.scope_id, u.scope_id, gm.member_id) as scoped_member_id,
+  'user' as type
+from
+  iam_group_member_user gm,
+  iam_user u,
+  iam_group g
+where
+  gm.member_id = u.public_id and
+  gm.group_id = g.public_id;
+  
+
 commit;
diff --git a/internal/iam/group_member.go b/internal/iam/group_member.go
new file mode 100644
index 000000000..7ffe1a4f3
--- /dev/null
+++ b/internal/iam/group_member.go
@@ -0,0 +1,110 @@
+package iam
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/hashicorp/watchtower/internal/db"
+	"github.com/hashicorp/watchtower/internal/iam/store"
+	"google.golang.org/protobuf/proto"
+)
+
+const (
+	groupMemberViewDefaultTableName = "iam_group_member"
+	groupMemberUserDefaultTable     = "iam_group_member_user"
+)
+
+// GroupMember provides a common way to return members.
+type GroupMember struct {
+	*store.GroupMemberView
+	tableName string `gorm:"-"`
+}
+
+// TableName provides an overridden gorm table name for group members.
+func (v *GroupMember) TableName() string {
+	if v.tableName != "" {
+		return v.tableName
+	}
+	return groupMemberViewDefaultTableName
+}
+
+// SetTableName sets the table name for the resource.  If the caller attempts to
+// set the name to "" the name will be reset to the default name.
+func (v *GroupMember) SetTableName(n string) {
+	switch n {
+	case "":
+		v.tableName = groupMemberViewDefaultTableName
+	default:
+		v.tableName = n
+	}
+}
+
+// GroupMemberUser is a group member that's a User
+type GroupMemberUser struct {
+	*store.GroupMemberUser
+	tableName string `gorm:"-"`
+}
+
+// ensure that GroupMember implements the interfaces of: Clonable, db.VetForWriter
+var _ Clonable = (*GroupMemberUser)(nil)
+var _ db.VetForWriter = (*GroupMemberUser)(nil)
+
+// NewGroupMemberUser creates a new in memory user member of the group. No
+// options are currently supported.
+func NewGroupMemberUser(groupId, userId string, opt ...Option) (*GroupMemberUser, error) {
+	if groupId == "" {
+		return nil, fmt.Errorf("new group member: missing group id: %w", db.ErrInvalidParameter)
+	}
+	if userId == "" {
+		return nil, fmt.Errorf("new group member: missing user id: %w", db.ErrInvalidParameter)
+	}
+	return &GroupMemberUser{
+		GroupMemberUser: &store.GroupMemberUser{
+			MemberId: userId,
+			GroupId:  groupId,
+		},
+	}, nil
+}
+
+func allocGroupMember() GroupMemberUser {
+	return GroupMemberUser{
+		GroupMemberUser: &store.GroupMemberUser{},
+	}
+}
+
+// Clone creates a clone of the GroupMember
+func (m *GroupMemberUser) Clone() interface{} {
+	cp := proto.Clone(m.GroupMemberUser)
+	return &GroupMemberUser{
+		GroupMemberUser: cp.(*store.GroupMemberUser),
+	}
+}
+
+// VetForWrite implements db.VetForWrite() interface for group members.
+func (m *GroupMemberUser) VetForWrite(ctx context.Context, r db.Reader, opType db.OpType, opt ...db.Option) error {
+	if m.GroupId == "" {
+		return fmt.Errorf("group member: missing group id: %w", db.ErrInvalidParameter)
+	}
+	if m.MemberId == "" {
+		return fmt.Errorf("group member: missing member id: %w", db.ErrInvalidParameter)
+	}
+	return nil
+}
+
+// TableName returns the tablename to override the default gorm table name
+func (m *GroupMemberUser) TableName() string {
+	if m.tableName != "" {
+		return m.tableName
+	}
+	return groupMemberUserDefaultTable
+}
+
+// SetTableName sets the tablename and satisfies the ReplayableMessage interface
+func (m *GroupMemberUser) SetTableName(n string) {
+	switch n {
+	case "":
+		m.tableName = groupMemberUserDefaultTable
+	default:
+		m.tableName = n
+	}
+}
diff --git a/internal/iam/group_member_test.go b/internal/iam/group_member_test.go
new file mode 100644
index 000000000..e6a60f07c
--- /dev/null
+++ b/internal/iam/group_member_test.go
@@ -0,0 +1,354 @@
+package iam
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/hashicorp/watchtower/internal/db"
+	"github.com/hashicorp/watchtower/internal/iam/store"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+)
+
+func Test_NewGroupMember(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	org, proj := TestScopes(t, conn)
+	org2, _ := TestScopes(t, conn)
+
+	orgGroup := TestGroup(t, conn, org.PublicId)
+	projGroup := TestGroup(t, conn, proj.PublicId)
+	user := TestUser(t, conn, org.PublicId)
+	user2 := TestUser(t, conn, org2.PublicId)
+
+	type args struct {
+		groupId string
+		userId  string
+		opt     []Option
+	}
+	tests := []struct {
+		name      string
+		args      args
+		want      *GroupMemberUser
+		wantErr   bool
+		wantIsErr error
+	}{
+		{
+			name: "valid-org",
+			args: args{
+				groupId: orgGroup.PublicId,
+				userId:  user.PublicId,
+			},
+			want: func() *GroupMemberUser {
+				gm := allocGroupMember()
+				gm.GroupId = orgGroup.PublicId
+				gm.MemberId = user.PublicId
+				return &gm
+			}(),
+		},
+		{
+			name: "valid-proj",
+			args: args{
+				groupId: projGroup.PublicId,
+				userId:  user.PublicId,
+			},
+			want: func() *GroupMemberUser {
+				gm := allocGroupMember()
+				gm.GroupId = projGroup.PublicId
+				gm.MemberId = user.PublicId
+				return &gm
+			}(),
+		},
+		{
+			name: "cross-org",
+			args: args{
+				groupId: orgGroup.PublicId,
+				userId:  user2.PublicId,
+			},
+			want: func() *GroupMemberUser {
+				gm := allocGroupMember()
+				gm.GroupId = orgGroup.PublicId
+				gm.MemberId = user2.PublicId
+				return &gm
+			}(),
+		},
+		{
+			name: "missing-group",
+			args: args{
+				userId: user.PublicId,
+			},
+			want:      nil,
+			wantErr:   true,
+			wantIsErr: db.ErrInvalidParameter,
+		},
+		{
+			name: "missing-user",
+			args: args{
+				groupId: projGroup.PublicId,
+			},
+			want:      nil,
+			wantErr:   true,
+			wantIsErr: db.ErrInvalidParameter,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert, require := assert.New(t), require.New(t)
+
+			got, err := NewGroupMemberUser(tt.args.groupId, tt.args.userId, tt.args.opt...)
+			if tt.wantErr {
+				require.Error(err)
+				assert.True(errors.Is(err, tt.wantIsErr))
+				return
+			}
+			require.NoError(err)
+			assert.Equal(tt.want, got)
+		})
+	}
+}
+
+func Test_GroupMemberCreate(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	org, proj := TestScopes(t, conn)
+	type args struct {
+		gm *GroupMemberUser
+	}
+	tests := []struct {
+		name       string
+		args       args
+		wantDup    bool
+		wantErr    bool
+		wantErrMsg string
+		wantIsErr  error
+	}{
+		{
+			name: "valid-with-org",
+			args: args{
+				gm: func() *GroupMemberUser {
+					g := TestGroup(t, conn, org.PublicId)
+					u := TestUser(t, conn, org.PublicId)
+					gm, err := NewGroupMemberUser(g.PublicId, u.PublicId)
+					require.NoError(t, err)
+					return gm
+				}(),
+			},
+			wantErr: false,
+		},
+		{
+			name: "valid-with-proj",
+			args: args{
+				gm: func() *GroupMemberUser {
+					g := TestGroup(t, conn, proj.PublicId)
+					u := TestUser(t, conn, org.PublicId)
+					gm, err := NewGroupMemberUser(g.PublicId, u.PublicId)
+					require.NoError(t, err)
+					return gm
+				}(),
+			},
+			wantErr: false,
+		},
+		{
+			name: "bad-group-id",
+			args: args{
+				gm: func() *GroupMemberUser {
+					id := testId(t)
+					u := TestUser(t, conn, org.PublicId)
+					gm, err := NewGroupMemberUser(id, u.PublicId)
+					require.NoError(t, err)
+					return gm
+				}(),
+			},
+			wantErr:    true,
+			wantErrMsg: `create: failed: pq: insert or update on table "iam_group_member_user" violates foreign key constraint`,
+		},
+		{
+			name: "bad-user-id",
+			args: args{
+				gm: func() *GroupMemberUser {
+					id := testId(t)
+					g := TestGroup(t, conn, proj.PublicId)
+					gm, err := NewGroupMemberUser(g.PublicId, id)
+					require.NoError(t, err)
+					return gm
+				}(),
+			},
+			wantErr:    true,
+			wantErrMsg: `create: failed: pq: insert or update on table "iam_group_member_user" violates foreign key constraint`,
+		},
+		{
+			name: "missing-group-id",
+			args: args{
+				gm: func() *GroupMemberUser {
+					u := TestUser(t, conn, org.PublicId)
+					return &GroupMemberUser{
+						GroupMemberUser: &store.GroupMemberUser{
+							GroupId:  "",
+							MemberId: u.PublicId,
+						},
+					}
+				}(),
+			},
+			wantErr:   true,
+			wantIsErr: db.ErrInvalidParameter,
+		},
+		{
+			name: "missing-user-id",
+			args: args{
+				gm: func() *GroupMemberUser {
+					g := TestGroup(t, conn, org.PublicId)
+					return &GroupMemberUser{
+						GroupMemberUser: &store.GroupMemberUser{
+							GroupId:  g.PublicId,
+							MemberId: "",
+						},
+					}
+				}(),
+			},
+			wantErr:   true,
+			wantIsErr: db.ErrInvalidParameter,
+		},
+		{
+			name: "dup-at-org",
+			args: args{
+				gm: func() *GroupMemberUser {
+					g := TestGroup(t, conn, org.PublicId)
+					u := TestUser(t, conn, org.PublicId)
+					gm, err := NewGroupMemberUser(g.PublicId, u.PublicId)
+					require.NoError(t, err)
+					return gm
+				}(),
+			},
+			wantDup:    true,
+			wantErr:    true,
+			wantErrMsg: `create: failed: pq: duplicate key value violates unique constraint "iam_group_member_user_pkey"`,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert, require := assert.New(t), require.New(t)
+			w := db.New(conn)
+			if tt.wantDup {
+				gm := tt.args.gm.Clone().(*GroupMemberUser)
+				err := w.Create(context.Background(), gm)
+				require.NoError(err)
+			}
+			gm := tt.args.gm.Clone().(*GroupMemberUser)
+			err := w.Create(context.Background(), gm)
+			if tt.wantErr {
+				require.Error(err)
+				assert.Contains(err.Error(), tt.wantErrMsg)
+				if tt.wantIsErr != nil {
+					assert.True(errors.Is(err, tt.wantIsErr))
+				}
+				return
+			}
+			assert.NoError(err)
+
+			found := allocGroupMember()
+			err = w.LookupWhere(context.Background(), &found, "group_id = ? and member_id = ?", gm.GroupId, gm.MemberId)
+			require.NoError(err)
+			assert.Equal(gm, &found)
+		})
+	}
+}
+
+func Test_GroupMemberUpdate(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	org, _ := TestScopes(t, conn)
+	rw := db.New(conn)
+
+	t.Run("updates not allowed", func(t *testing.T) {
+		assert, require := assert.New(t), require.New(t)
+		g := TestGroup(t, conn, org.PublicId)
+		u := TestUser(t, conn, org.PublicId)
+		u2 := TestUser(t, conn, org.PublicId)
+		gm := TestGroupMember(t, conn, g.PublicId, u.PublicId)
+		updateGrpMember := gm.Clone().(*GroupMemberUser)
+		updateGrpMember.MemberId = u2.PublicId
+		updatedRows, err := rw.Update(context.Background(), updateGrpMember, []string{"MemberId"}, nil)
+		require.Error(err)
+		assert.Equal(0, updatedRows)
+	})
+}
+
+func Test_GroupMemberDelete(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	id := testId(t)
+	org, _ := TestScopes(t, conn)
+	u := TestUser(t, conn, org.PublicId)
+	g := TestGroup(t, conn, org.PublicId)
+
+	tests := []struct {
+		name            string
+		gm              *GroupMemberUser
+		wantRowsDeleted int
+		wantErr         bool
+		wantErrMsg      string
+	}{
+		{
+			name:            "valid",
+			gm:              TestGroupMember(t, conn, g.PublicId, u.PublicId),
+			wantErr:         false,
+			wantRowsDeleted: 1,
+		},
+		{
+			name:            "bad-id",
+			gm:              func() *GroupMemberUser { gm := allocGroupMember(); gm.MemberId = id; gm.GroupId = id; return &gm }(),
+			wantErr:         false,
+			wantRowsDeleted: 0,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert, require := assert.New(t), require.New(t)
+			deleteGroup := allocGroupMember()
+			deleteGroup.GroupId = tt.gm.GetGroupId()
+			deleteGroup.MemberId = tt.gm.GetMemberId()
+			deletedRows, err := rw.Delete(context.Background(), &deleteGroup)
+			if tt.wantErr {
+				require.Error(err)
+				return
+			}
+			require.NoError(err)
+			if tt.wantRowsDeleted == 0 {
+				assert.Equal(tt.wantRowsDeleted, deletedRows)
+				return
+			}
+			assert.Equal(tt.wantRowsDeleted, deletedRows)
+			found := allocGroupMember()
+			err = rw.LookupWhere(context.Background(), &found, "group_id = ? and member_id = ?", tt.gm.GetGroupId(), tt.gm.GetMemberId())
+			require.Error(err)
+			assert.True(errors.Is(db.ErrRecordNotFound, err))
+		})
+	}
+}
+
+func TestGroupMember_Clone(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	org, proj := TestScopes(t, conn)
+	user := TestUser(t, conn, org.PublicId)
+	t.Run("valid", func(t *testing.T) {
+		assert := assert.New(t)
+		group := TestGroup(t, conn, org.PublicId)
+		gm := TestGroupMember(t, conn, group.PublicId, user.PublicId)
+		cp := gm.Clone()
+		assert.True(proto.Equal(cp.(*GroupMemberUser).GroupMemberUser, gm.GroupMemberUser))
+	})
+	t.Run("not-equal", func(t *testing.T) {
+		assert := assert.New(t)
+		g := TestGroup(t, conn, org.PublicId)
+		g2 := TestGroup(t, conn, proj.PublicId)
+		gm := TestGroupMember(t, conn, g.PublicId, user.PublicId)
+		gm2 := TestGroupMember(t, conn, g2.PublicId, user.PublicId)
+		cp := gm.Clone()
+		assert.True(!proto.Equal(cp.(*GroupMemberUser).GroupMemberUser, gm2.GroupMemberUser))
+	})
+}
diff --git a/internal/iam/repository_group.go b/internal/iam/repository_group.go
index 58f4480ff..78fac9d2c 100644
--- a/internal/iam/repository_group.go
+++ b/internal/iam/repository_group.go
@@ -6,6 +6,7 @@ import (
 	"strings"
 
 	"github.com/hashicorp/watchtower/internal/db"
+	"github.com/hashicorp/watchtower/internal/oplog"
 )
 
 // CreateGroup will create a group in the repository and return the written
@@ -127,3 +128,313 @@ func (r *Repository) ListGroups(ctx context.Context, withScopeId string, opt ...
 	}
 	return grps, nil
 }
+
+// ListGroupMembers of a group and supports WithLimit option.
+func (r *Repository) ListGroupMembers(ctx context.Context, withGroupId string, opt ...Option) ([]*GroupMember, error) {
+	if withGroupId == "" {
+		return nil, fmt.Errorf("list group members: missing group id: %w", db.ErrInvalidParameter)
+	}
+	members := []*GroupMember{}
+	if err := r.list(ctx, &members, "group_id = ?", []interface{}{withGroupId}, opt...); err != nil {
+		return nil, fmt.Errorf("list group members: %w", err)
+	}
+	return members, nil
+}
+
+// AddGroupMembers provides the ability to add members (userIds) to a group
+// (groupId).  The group's current db version must match the groupVersion or an
+// error will be returned.
+func (r *Repository) AddGroupMembers(ctx context.Context, groupId string, groupVersion uint32, userIds []string, opt ...Option) ([]*GroupMember, error) {
+
+	if groupId == "" {
+		return nil, fmt.Errorf("add group members: missing group id %w", db.ErrInvalidParameter)
+	}
+	if len(userIds) == 0 {
+		return nil, fmt.Errorf("add group members: missing user ids to add %w", db.ErrInvalidParameter)
+	}
+
+	group := allocGroup()
+	group.PublicId = groupId
+	scope, err := group.GetScope(ctx, r.reader)
+	if err != nil {
+		return nil, fmt.Errorf("add group members: unable to get group %s scope: %w", groupId, err)
+	}
+
+	newGroupMembers := make([]interface{}, 0, len(userIds))
+	for _, id := range userIds {
+		gm, err := NewGroupMemberUser(groupId, id)
+		if err != nil {
+			return nil, fmt.Errorf("add group members: unable to create in memory group member: %w", err)
+		}
+		newGroupMembers = append(newGroupMembers, gm)
+	}
+
+	var currentMembers []*GroupMember
+	_, err = r.writer.DoTx(
+		ctx,
+		db.StdRetryCnt,
+		db.ExpBackoff{},
+		func(reader db.Reader, w db.Writer) error {
+			msgs := make([]*oplog.Message, 0, 2)
+			groupTicket, err := w.GetTicket(&group)
+			if err != nil {
+				return fmt.Errorf("add group members: unable to get ticket: %w", err)
+			}
+			updatedGroup := allocGroup()
+			updatedGroup.PublicId = groupId
+			updatedGroup.Version = uint32(groupVersion) + 1
+			var groupOplogMsg oplog.Message
+			rowsUpdated, err := w.Update(ctx, &updatedGroup, []string{"Version"}, nil, db.NewOplogMsg(&groupOplogMsg), db.WithVersion(groupVersion))
+			if err != nil {
+				return fmt.Errorf("add group members: unable to update group version: %w", err)
+			}
+			if rowsUpdated != 1 {
+				return fmt.Errorf("add group members: updated group and %d rows updated", rowsUpdated)
+			}
+			msgs = append(msgs, &groupOplogMsg)
+			memberOplogMsgs := make([]*oplog.Message, 0, len(newGroupMembers))
+			if err := w.CreateItems(ctx, newGroupMembers, db.NewOplogMsgs(&memberOplogMsgs)); err != nil {
+				return fmt.Errorf("add group members: unable to add users: %w", err)
+			}
+			msgs = append(msgs, memberOplogMsgs...)
+			metadata := oplog.Metadata{
+				"op-type":            []string{oplog.OpType_OP_TYPE_CREATE.String()},
+				"scope-id":           []string{scope.PublicId},
+				"scope-type":         []string{scope.Type},
+				"resource-public-id": []string{groupId},
+			}
+			if err := w.WriteOplogEntryWith(ctx, r.wrapper, groupTicket, metadata, msgs); err != nil {
+				return fmt.Errorf("add group members: 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,
+				wrapper: r.wrapper,
+				// intentionally not setting the defaultLimit, so we'll get all
+				// the members without a limit
+			}
+			currentMembers, err = txRepo.ListGroupMembers(ctx, groupId)
+			if err != nil {
+				return fmt.Errorf("set group members: unable to retrieve current group members after sets: %w", err)
+			}
+			return nil
+		},
+	)
+	if err != nil {
+		return nil, fmt.Errorf("add group members: error adding members: %w", err)
+	}
+	return currentMembers, nil
+}
+
+// DeleteGroupMembers (userIds) from a group (groupId). The group's current db version
+// must match the groupVersion or an error will be returned.
+func (r *Repository) DeleteGroupMembers(ctx context.Context, groupId string, groupVersion uint32, userIds []string, opt ...Option) (int, error) {
+	if groupId == "" {
+		return db.NoRowsAffected, fmt.Errorf("delete group members: missing group id: %w", db.ErrInvalidParameter)
+	}
+	if len(userIds) == 0 {
+		return db.NoRowsAffected, fmt.Errorf("delete group members: missing either user or groups to delete %w", db.ErrInvalidParameter)
+	}
+	group := allocGroup()
+	group.PublicId = groupId
+	scope, err := group.GetScope(ctx, r.reader)
+	if err != nil {
+		return db.NoRowsAffected, fmt.Errorf("delete group members: unable to get group %s scope: %w", groupId, err)
+	}
+
+	deleteMembers := make([]interface{}, 0, len(userIds))
+	for _, id := range userIds {
+		member, err := NewGroupMemberUser(groupId, id)
+		if err != nil {
+			return db.NoRowsAffected, fmt.Errorf("delete group members: unable to create in memory group member: %w", err)
+		}
+		deleteMembers = append(deleteMembers, member)
+	}
+
+	var totalRowsDeleted int
+	_, err = r.writer.DoTx(
+		ctx,
+		db.StdRetryCnt,
+		db.ExpBackoff{},
+		func(reader db.Reader, w db.Writer) error {
+			msgs := make([]*oplog.Message, 0, 2)
+			groupTicket, err := w.GetTicket(&group)
+			if err != nil {
+				return fmt.Errorf("delete group members: unable to get ticket: %w", err)
+			}
+			updatedGroup := allocGroup()
+			updatedGroup.PublicId = groupId
+			updatedGroup.Version = uint32(groupVersion) + 1
+			var groupOplogMsg oplog.Message
+			rowsUpdated, err := w.Update(ctx, &updatedGroup, []string{"Version"}, nil, db.NewOplogMsg(&groupOplogMsg), db.WithVersion(groupVersion))
+			if err != nil {
+				return fmt.Errorf("delete group members: unable to update group version: %w", err)
+			}
+			if rowsUpdated != 1 {
+				return fmt.Errorf("delete group members: updated group and %d rows updated", rowsUpdated)
+			}
+			msgs = append(msgs, &groupOplogMsg)
+			userOplogMsgs := make([]*oplog.Message, 0, len(deleteMembers))
+			rowsDeleted, err := w.DeleteItems(ctx, deleteMembers, db.NewOplogMsgs(&userOplogMsgs))
+			if err != nil {
+				return fmt.Errorf("delete group members: unable to delete group members: %w", err)
+			}
+			if rowsDeleted != len(deleteMembers) {
+				return fmt.Errorf("delete group members: group members deleted %d did not match request for %d", rowsDeleted, len(deleteMembers))
+			}
+			totalRowsDeleted += rowsDeleted
+			msgs = append(msgs, userOplogMsgs...)
+			metadata := oplog.Metadata{
+				"op-type":            []string{oplog.OpType_OP_TYPE_DELETE.String()},
+				"scope-id":           []string{scope.PublicId},
+				"scope-type":         []string{scope.Type},
+				"resource-public-id": []string{groupId},
+			}
+			if err := w.WriteOplogEntryWith(ctx, r.wrapper, groupTicket, metadata, msgs); err != nil {
+				return fmt.Errorf("delete group members: unable to write oplog: %w", err)
+			}
+			return nil
+		},
+	)
+	if err != nil {
+		return db.NoRowsAffected, fmt.Errorf("delete group members: error deleting members: %w", err)
+	}
+	return totalRowsDeleted, nil
+}
+
+// SetGroupMembers will set the group's members.  If userIds is empty, the
+// members will be cleared.
+func (r *Repository) SetGroupMembers(ctx context.Context, groupId string, groupVersion uint32, userIds []string, opt ...Option) ([]*GroupMember, int, error) {
+	if groupId == "" {
+		return nil, db.NoRowsAffected, fmt.Errorf("set group members: missing role id: %w", db.ErrInvalidParameter)
+	}
+	group := allocGroup()
+	group.PublicId = groupId
+	scope, err := group.GetScope(ctx, r.reader)
+	if err != nil {
+		return nil, db.NoRowsAffected, fmt.Errorf("set group members: unable to get role %s scope: %w", groupId, err)
+	}
+
+	// find existing members (since we're using groupVersion, we can safely do
+	// this here, outside the TxHandler)
+	currentMembers := []*GroupMember{}
+	if err := r.reader.SearchWhere(ctx, &currentMembers, "group_id = ?", []interface{}{groupId}); err != nil {
+		return nil, db.NoRowsAffected, fmt.Errorf("set group members: unable to search for existing members of group %s: %w", groupId, err)
+	}
+	found := map[string]*GroupMember{}
+	for _, m := range currentMembers {
+		found[m.GroupId+m.MemberId] = m
+	}
+	addMembers := make([]interface{}, 0, len(userIds))
+	deleteMembers := make([]interface{}, 0, len(userIds))
+
+	for _, usrId := range userIds {
+		_, ok := found[groupId+usrId]
+		if ok {
+			// we have a match, so do nada since we want to keep it, but remove
+			// it from found.
+			delete(found, groupId+usrId)
+			continue
+		}
+		// not found, so we add it
+		gm, err := NewGroupMemberUser(groupId, usrId)
+		if err != nil {
+			return nil, db.NoRowsAffected, fmt.Errorf("set group members: unable to create in memory group member: %w", err)
+		}
+		addMembers = append(addMembers, gm)
+	}
+	if len(found) > 0 {
+		for _, fgm := range found {
+			// not found, so we add it
+			gm, err := NewGroupMemberUser(fgm.GroupId, fgm.MemberId)
+			if err != nil {
+				return nil, db.NoRowsAffected, fmt.Errorf("set group members: unable to create in memory group member: %w", err)
+			}
+			deleteMembers = append(deleteMembers, gm)
+		}
+	}
+
+	// handle no change to existing group members
+	if len(addMembers) == 0 && len(deleteMembers) == 0 {
+		return currentMembers, db.NoRowsAffected, nil
+	}
+
+	var totalRowsAffected int
+	_, err = r.writer.DoTx(
+		ctx,
+		db.StdRetryCnt,
+		db.ExpBackoff{},
+		func(reader db.Reader, w db.Writer) error {
+			msgs := make([]*oplog.Message, 0, 2)
+			metadata := oplog.Metadata{
+				"op-type":            []string{oplog.OpType_OP_TYPE_UPDATE.String()},
+				"scope-id":           []string{scope.PublicId},
+				"scope-type":         []string{scope.Type},
+				"resource-public-id": []string{groupId},
+			}
+			// we need a group, which won't be redeemed until all the other
+			// writes are successful.  We can't just use a single ticket because
+			// we need to write oplog entries for deletes and adds
+			groupTicket, err := w.GetTicket(&group)
+			if err != nil {
+				return fmt.Errorf("set group members: unable to get ticket for group: %w", err)
+			}
+			updatedGroup := allocGroup()
+			updatedGroup.PublicId = groupId
+			updatedGroup.Version = uint32(groupVersion) + 1
+			var groupOplogMsg oplog.Message
+			rowsUpdated, err := w.Update(ctx, &updatedGroup, []string{"Version"}, nil, db.NewOplogMsg(&groupOplogMsg), db.WithVersion(groupVersion))
+			if err != nil {
+				return fmt.Errorf("set group members: unable to update group version: %w", err)
+			}
+			if rowsUpdated != 1 {
+				return fmt.Errorf("set group members: updated group and %d rows updated", rowsUpdated)
+			}
+			if len(deleteMembers) > 0 {
+				userOplogMsgs := make([]*oplog.Message, 0, len(deleteMembers))
+				rowsDeleted, err := w.DeleteItems(ctx, deleteMembers, db.NewOplogMsgs(&userOplogMsgs))
+				if err != nil {
+					return fmt.Errorf("set group members: unable to delete user roles: %w", err)
+				}
+				if rowsDeleted != len(deleteMembers) {
+					return fmt.Errorf("set group members: members deleted %d did not match request for %d", rowsDeleted, len(deleteMembers))
+				}
+				totalRowsAffected += rowsDeleted
+				msgs = append(msgs, userOplogMsgs...)
+				metadata["op-type"] = append(metadata["op-type"], oplog.OpType_OP_TYPE_DELETE.String())
+			}
+			if len(addMembers) > 0 {
+				userOplogMsgs := make([]*oplog.Message, 0, len(addMembers))
+				if err := w.CreateItems(ctx, addMembers, db.NewOplogMsgs(&userOplogMsgs)); err != nil {
+					return fmt.Errorf("set group members: unable to add users: %w", err)
+				}
+				totalRowsAffected += len(addMembers)
+				msgs = append(msgs, userOplogMsgs...)
+				metadata["op-type"] = append(metadata["op-type"], oplog.OpType_OP_TYPE_CREATE.String())
+
+			}
+			// we're done with all the principal writes, so let's write the
+			// role's update oplog message
+			if err := w.WriteOplogEntryWith(ctx, r.wrapper, groupTicket, metadata, msgs); err != nil {
+				return fmt.Errorf("set group members: unable to write oplog for additions: %w", err)
+			}
+			// we need a new repo, that's using the same reader/writer as this TxHandler
+			txRepo := Repository{
+				reader:  reader,
+				writer:  w,
+				wrapper: r.wrapper,
+				// intentionally not setting the defaultLimit, so we'll get all
+				// the members without a limit
+			}
+			currentMembers, err = txRepo.ListGroupMembers(ctx, groupId)
+			if err != nil {
+				return fmt.Errorf("set group members: unable to retrieve current group members after sets: %w", err)
+			}
+			return nil
+		})
+	if err != nil {
+		return nil, db.NoRowsAffected, fmt.Errorf("set group members: unable to set group members: %w", err)
+	}
+	return currentMembers, totalRowsAffected, nil
+}
diff --git a/internal/iam/repository_group_test.go b/internal/iam/repository_group_test.go
index 10e76cd6f..18ddefdea 100644
--- a/internal/iam/repository_group_test.go
+++ b/internal/iam/repository_group_test.go
@@ -3,6 +3,7 @@ package iam
 import (
 	"context"
 	"errors"
+	"sort"
 	"testing"
 	"time"
 
@@ -644,3 +645,441 @@ func TestRepository_ListGroups(t *testing.T) {
 		})
 	}
 }
+
+func TestRepository_ListMembers(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	const testLimit = 10
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	repo, err := NewRepository(rw, rw, wrapper, WithLimit(testLimit))
+	require.NoError(t, err)
+	org, proj := TestScopes(t, conn)
+	pg := TestGroup(t, conn, proj.PublicId)
+	og := TestGroup(t, conn, org.PublicId)
+
+	type args struct {
+		withGroupId string
+		opt         []Option
+	}
+	tests := []struct {
+		name      string
+		createCnt int
+		args      args
+		wantCnt   int
+		wantErr   bool
+	}{
+		{
+			name:      "no-limit-pg-group",
+			createCnt: repo.defaultLimit + 1,
+			args: args{
+				withGroupId: pg.PublicId,
+				opt:         []Option{WithLimit(-1)},
+			},
+			wantCnt: repo.defaultLimit + 1,
+			wantErr: false,
+		},
+		{
+			name:      "no-limit-org-group",
+			createCnt: repo.defaultLimit + 1,
+			args: args{
+				withGroupId: og.PublicId,
+				opt:         []Option{WithLimit(-1)},
+			},
+			wantCnt: repo.defaultLimit + 1,
+			wantErr: false,
+		},
+		{
+			name:      "default-limit",
+			createCnt: repo.defaultLimit + 1,
+			args: args{
+				withGroupId: pg.PublicId,
+			},
+			wantCnt: repo.defaultLimit,
+			wantErr: false,
+		},
+		{
+			name:      "custom-limit",
+			createCnt: repo.defaultLimit + 1,
+			args: args{
+				withGroupId: pg.PublicId,
+				opt:         []Option{WithLimit(3)},
+			},
+			wantCnt: 3,
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert, require := assert.New(t), require.New(t)
+			require.NoError(conn.Where("1=1").Delete(allocGroupMember()).Error)
+			gm := []*GroupMemberUser{}
+			for i := 0; i < tt.createCnt; i++ {
+				u := TestUser(t, conn, org.PublicId)
+				gm = append(gm, TestGroupMember(t, conn, tt.args.withGroupId, u.PublicId))
+			}
+			assert.Equal(tt.createCnt, len(gm))
+
+			got, err := repo.ListGroupMembers(context.Background(), tt.args.withGroupId, tt.args.opt...)
+			if tt.wantErr {
+				require.Error(err)
+				return
+			}
+			require.NoError(err)
+			assert.Equal(tt.wantCnt, len(got))
+		})
+	}
+	t.Run("missing-id", func(t *testing.T) {
+		require := require.New(t)
+		got, err := repo.ListGroupMembers(context.Background(), "")
+		require.Error(err)
+		require.Nil(got)
+		require.Truef(errors.Is(err, db.ErrInvalidParameter), "unexpected error %s", err.Error())
+
+	})
+}
+
+func TestRepository_AddGroupMembers(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	repo, err := NewRepository(rw, rw, wrapper)
+	require.NoError(t, err)
+	org, proj := TestScopes(t, conn)
+	group := TestGroup(t, conn, proj.PublicId)
+	createUsersFn := func() []string {
+		results := []string{}
+		for i := 0; i < 5; i++ {
+			u := TestUser(t, conn, org.PublicId)
+			results = append(results, u.PublicId)
+		}
+		return results
+	}
+	groupVersion := uint32(0)
+	type args struct {
+		groupId      string
+		groupVersion uint32
+		userIds      []string
+		opt          []Option
+	}
+	tests := []struct {
+		name      string
+		args      args
+		wantErr   bool
+		wantErrIs error
+	}{
+		{
+			name: "valid-members",
+			args: args{
+				groupId: group.PublicId,
+				userIds: createUsersFn(),
+			},
+			wantErr: false,
+		},
+		{
+			name: "valid-next-version",
+			args: args{
+				groupId: group.PublicId,
+				userIds: createUsersFn(),
+			},
+			wantErr: false,
+		},
+		{
+			name: "bad-version",
+			args: args{
+				groupId:      group.PublicId,
+				groupVersion: 1000,
+				userIds:      createUsersFn(),
+			},
+			wantErr: true,
+		},
+		{
+			name: "no-members",
+			args: args{
+				groupId: group.PublicId,
+				userIds: nil,
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert, require := assert.New(t), require.New(t)
+
+			var version uint32
+			if tt.args.groupVersion == 0 {
+				groupVersion += 1
+				version = groupVersion
+			} else {
+				version = tt.args.groupVersion
+			}
+			require.NoError(conn.Where("1=1").Delete(allocGroupMember()).Error)
+			got, err := repo.AddGroupMembers(context.Background(), tt.args.groupId, version, tt.args.userIds, 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)
+			gotMembers := map[string]*GroupMember{}
+			for _, m := range got {
+				gotMembers[m.MemberId] = m
+			}
+			for _, id := range tt.args.userIds {
+				assert.NotEmpty(gotMembers[id])
+				u, err := repo.LookupUser(context.Background(), id)
+				assert.NoError(err)
+				assert.Equal(id, u.PublicId)
+			}
+			err = db.TestVerifyOplog(t, rw, group.PublicId, db.WithOperation(oplog.OpType_OP_TYPE_CREATE), db.WithCreateNotBefore(10*time.Second))
+			assert.NoError(err)
+
+			foundMembers, err := repo.ListGroupMembers(context.Background(), group.PublicId)
+			require.NoError(err)
+			for _, m := range foundMembers {
+				assert.NotEmpty(gotMembers[m.MemberId])
+				assert.Equal(gotMembers[m.MemberId].GetGroupId(), m.GroupId)
+			}
+
+		})
+	}
+}
+
+func TestRepository_DeleteGroupMembers(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	repo, err := NewRepository(rw, rw, wrapper)
+	require.NoError(t, err)
+	org, _ := TestScopes(t, conn)
+
+	type args struct {
+		group           *Group
+		groupIdOverride *string
+		groupVersion    uint32
+		createUserCnt   int
+		deleteUserCnt   int
+		opt             []Option
+	}
+	tests := []struct {
+		name            string
+		args            args
+		wantRowsDeleted int
+		wantErr         bool
+		wantIsErr       error
+	}{
+		{
+			name: "valid",
+			args: args{
+				group:         TestGroup(t, conn, org.PublicId),
+				createUserCnt: 5,
+				deleteUserCnt: 5,
+				groupVersion:  2,
+			},
+			wantRowsDeleted: 5,
+			wantErr:         false,
+		},
+		{
+			name: "valid-keeping-some",
+			args: args{
+				group:         TestGroup(t, conn, org.PublicId),
+				createUserCnt: 5,
+				deleteUserCnt: 2,
+				groupVersion:  2,
+			},
+			wantRowsDeleted: 2,
+			wantErr:         false,
+		},
+		{
+			name: "not-found",
+			args: args{
+				group:           TestGroup(t, conn, org.PublicId),
+				groupVersion:    2,
+				groupIdOverride: func() *string { id := testId(t); return &id }(),
+				createUserCnt:   5,
+				deleteUserCnt:   5,
+			},
+			wantRowsDeleted: 0,
+			wantErr:         true,
+		},
+		{
+			name: "missing-group-id",
+			args: args{
+				group:           TestGroup(t, conn, org.PublicId),
+				groupVersion:    2,
+				groupIdOverride: func() *string { id := ""; return &id }(),
+				createUserCnt:   5,
+				deleteUserCnt:   5,
+			},
+			wantRowsDeleted: 0,
+			wantErr:         true,
+			wantIsErr:       db.ErrInvalidParameter,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert, require := assert.New(t), require.New(t)
+			userIds := make([]string, 0, tt.args.createUserCnt)
+			for i := 0; i < tt.args.createUserCnt; i++ {
+				u := TestUser(t, conn, org.PublicId)
+				userIds = append(userIds, u.PublicId)
+			}
+			members, err := repo.AddGroupMembers(context.Background(), tt.args.group.PublicId, 1, userIds, tt.args.opt...)
+			require.NoError(err)
+			assert.Equal(tt.args.createUserCnt, len(members))
+
+			deleteUserIds := make([]string, 0, tt.args.deleteUserCnt)
+			for i := 0; i < tt.args.deleteUserCnt; i++ {
+				deleteUserIds = append(deleteUserIds, userIds[i])
+			}
+			var groupId string
+			switch {
+			case tt.args.groupIdOverride != nil:
+				groupId = *tt.args.groupIdOverride
+			default:
+				groupId = tt.args.group.PublicId
+			}
+			deletedRows, err := repo.DeleteGroupMembers(context.Background(), groupId, tt.args.groupVersion, deleteUserIds, tt.args.opt...)
+			if tt.wantErr {
+				assert.Error(err)
+				assert.Equal(0, deletedRows)
+				if tt.wantIsErr != nil {
+					assert.Truef(errors.Is(err, tt.wantIsErr), "unexpected error %s", err.Error())
+				}
+				err = db.TestVerifyOplog(t, rw, tt.args.group.PublicId, db.WithOperation(oplog.OpType_OP_TYPE_DELETE), db.WithCreateNotBefore(10*time.Second))
+				assert.Error(err)
+				assert.True(errors.Is(db.ErrRecordNotFound, err))
+				return
+			}
+			require.NoError(err)
+			assert.Equal(tt.wantRowsDeleted, deletedRows)
+
+			err = db.TestVerifyOplog(t, rw, tt.args.group.PublicId, db.WithOperation(oplog.OpType_OP_TYPE_DELETE), db.WithCreateNotBefore(10*time.Second))
+			assert.NoError(err)
+		})
+	}
+}
+
+func TestRepository_SetGroupMembers(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+
+	repo, err := NewRepository(rw, rw, wrapper)
+	require.NoError(t, err)
+
+	org, proj := TestScopes(t, conn)
+	testUser := TestUser(t, conn, org.PublicId)
+
+	createUsersFn := func() []string {
+		results := []string{}
+		for i := 0; i < 5; i++ {
+			u := TestUser(t, conn, org.PublicId)
+			results = append(results, u.PublicId)
+		}
+		return results
+	}
+	setupFn := func(groupId string) []string {
+		users := createUsersFn()
+		_, err := repo.AddGroupMembers(context.Background(), groupId, 1, users)
+		require.NoError(t, err)
+		return users
+	}
+	type args struct {
+		group          *Group
+		groupVersion   uint32
+		userIds        []string
+		addToOrigUsers bool
+		opt            []Option
+	}
+	tests := []struct {
+		name             string
+		setup            func(string) []string
+		args             args
+		wantAffectedRows int
+		wantErr          bool
+	}{
+		{
+			name:  "clear",
+			setup: setupFn,
+			args: args{
+				group:        TestGroup(t, conn, proj.PublicId),
+				groupVersion: 2, // yep, since setupFn will increment it to 2
+				userIds:      []string{},
+			},
+			wantErr:          false,
+			wantAffectedRows: 5,
+		},
+		{
+			name:  "no change",
+			setup: setupFn,
+			args: args{
+				group:          TestGroup(t, conn, proj.PublicId),
+				groupVersion:   2, // yep, since setupFn will increment it to 2
+				userIds:        []string{},
+				addToOrigUsers: true,
+			},
+			wantErr:          false,
+			wantAffectedRows: 0,
+		},
+		{
+			name:  "add users",
+			setup: setupFn,
+			args: args{
+				group:          TestGroup(t, conn, proj.PublicId),
+				groupVersion:   2, // yep, since setupFn will increment it to 2
+				userIds:        []string{testUser.PublicId},
+				addToOrigUsers: true,
+			},
+			wantErr:          false,
+			wantAffectedRows: 1,
+		},
+		{
+			name:  "remove existing and add users",
+			setup: setupFn,
+			args: args{
+				group:          TestGroup(t, conn, proj.PublicId),
+				groupVersion:   2, // yep, since setupFn will increment it to 2
+				userIds:        []string{testUser.PublicId},
+				addToOrigUsers: false,
+			},
+			wantErr:          false,
+			wantAffectedRows: 6,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert, require := assert.New(t), require.New(t)
+			var origUsers []string
+			if tt.setup != nil {
+				origUsers = tt.setup(tt.args.group.PublicId)
+			}
+			setUsers := tt.args.userIds
+			if tt.args.addToOrigUsers {
+				setUsers = append(setUsers, origUsers...)
+			}
+
+			got, affectedRows, err := repo.SetGroupMembers(context.Background(), tt.args.group.PublicId, tt.args.groupVersion, setUsers, tt.args.opt...)
+			if tt.wantErr {
+				require.Error(err)
+				return
+			}
+			require.NoError(err)
+			assert.Equal(tt.wantAffectedRows, affectedRows)
+			var gotIds []string
+			for _, r := range got {
+				gotIds = append(gotIds, r.GetMemberId())
+			}
+			var wantIds []string
+			wantIds = append(wantIds, tt.args.userIds...)
+			sort.Strings(wantIds)
+			sort.Strings(gotIds)
+			assert.Equal(wantIds, wantIds)
+		})
+	}
+}
diff --git a/internal/iam/repository_user_test.go b/internal/iam/repository_user_test.go
index 650e4529f..6aed43fc8 100644
--- a/internal/iam/repository_user_test.go
+++ b/internal/iam/repository_user_test.go
@@ -339,7 +339,7 @@ func TestRepository_UpdateUser(t *testing.T) {
 				case "dup-name":
 					assert.Equal(fmt.Sprintf(tt.wantErrMsg, "dup-name"+id, org.PublicId), err.Error())
 				default:
-					assert.True(strings.Contains(err.Error(), tt.wantErrMsg), "%s", err.Error())
+					assert.Containsf(err.Error(), tt.wantErrMsg, "unexpected error: %s", err.Error())
 				}
 				err = db.TestVerifyOplog(t, rw, u.PublicId, db.WithOperation(oplog.OpType_OP_TYPE_UPDATE), db.WithCreateNotBefore(10*time.Second))
 				require.Error(err)
diff --git a/internal/iam/store/group.pb.go b/internal/iam/store/group.pb.go
index 6828ea936..c79ce3dfc 100644
--- a/internal/iam/store/group.pb.go
+++ b/internal/iam/store/group.pb.go
@@ -53,6 +53,10 @@ type Group struct {
 	// disabled is by default false and allows a Group to be marked disabled.
 	// @inject_tag: `gorm:"default:null"`
 	Disabled bool `protobuf:"varint,7,opt,name=disabled,proto3" json:"disabled,omitempty" gorm:"default:null"`
+	// version allows optimistic locking of the group when modifying the group
+	// itself and when modifying dependent items like group members.
+	// @inject_tag: `gorm:"default:null"`
+	Version uint32 `protobuf:"varint,8,opt,name=version,proto3" json:"version,omitempty" gorm:"default:null"`
 }
 
 func (x *Group) Reset() {
@@ -136,6 +140,13 @@ func (x *Group) GetDisabled() bool {
 	return false
 }
 
+func (x *Group) GetVersion() uint32 {
+	if x != nil {
+		return x.Version
+	}
+	return 0
+}
+
 var File_controller_storage_iam_store_v1_group_proto protoreflect.FileDescriptor
 
 var file_controller_storage_iam_store_v1_group_proto_rawDesc = []byte{
@@ -146,31 +157,35 @@ var file_controller_storage_iam_store_v1_group_proto_rawDesc = []byte{
 	0x65, 0x2e, 0x69, 0x61, 0x6d, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x1a, 0x2f,
 	0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x61,
 	0x67, 0x65, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2f, 0x76, 0x31, 0x2f,
-	0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22,
-	0xab, 0x02, 0x0a, 0x05, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x75, 0x62,
-	0x6c, 0x69, 0x63, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x75,
-	0x62, 0x6c, 0x69, 0x63, 0x49, 0x64, 0x12, 0x4b, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65,
-	0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f,
-	0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65,
-	0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69,
-	0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x54,
-	0x69, 0x6d, 0x65, 0x12, 0x4b, 0x0a, 0x0b, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69,
-	0x6d, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72,
+	0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a,
+	0x2b, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f, 0x72,
+	0x61, 0x67, 0x65, 0x2f, 0x69, 0x61, 0x6d, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x2f, 0x76, 0x31,
+	0x2f, 0x73, 0x63, 0x6f, 0x70, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xc5, 0x02, 0x0a,
+	0x05, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63,
+	0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x75, 0x62, 0x6c, 0x69,
+	0x63, 0x49, 0x64, 0x12, 0x4b, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69,
+	0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72,
 	0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69,
 	0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73,
-	0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65,
-	0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04,
-	0x6e, 0x61, 0x6d, 0x65, 0x12, 0x20, 0x0a, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74,
-	0x69, 0x6f, 0x6e, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72,
-	0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x19, 0x0a, 0x08, 0x73, 0x63, 0x6f, 0x70, 0x65, 0x5f,
-	0x69, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x73, 0x63, 0x6f, 0x70, 0x65, 0x49,
-	0x64, 0x12, 0x1a, 0x0a, 0x08, 0x64, 0x69, 0x73, 0x61, 0x62, 0x6c, 0x65, 0x64, 0x18, 0x07, 0x20,
-	0x01, 0x28, 0x08, 0x52, 0x08, 0x64, 0x69, 0x73, 0x61, 0x62, 0x6c, 0x65, 0x64, 0x42, 0x3a, 0x5a,
-	0x38, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68,
-	0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x77, 0x61, 0x74, 0x63, 0x68, 0x74, 0x6f, 0x77, 0x65, 0x72,
-	0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x69, 0x61, 0x6d, 0x2f, 0x73, 0x74,
-	0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f,
-	0x33,
+	0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65,
+	0x12, 0x4b, 0x0a, 0x0b, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18,
+	0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c,
+	0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73,
+	0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d,
+	0x70, 0x52, 0x0a, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x12, 0x0a,
+	0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d,
+	0x65, 0x12, 0x20, 0x0a, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e,
+	0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74,
+	0x69, 0x6f, 0x6e, 0x12, 0x19, 0x0a, 0x08, 0x73, 0x63, 0x6f, 0x70, 0x65, 0x5f, 0x69, 0x64, 0x18,
+	0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x73, 0x63, 0x6f, 0x70, 0x65, 0x49, 0x64, 0x12, 0x1a,
+	0x0a, 0x08, 0x64, 0x69, 0x73, 0x61, 0x62, 0x6c, 0x65, 0x64, 0x18, 0x07, 0x20, 0x01, 0x28, 0x08,
+	0x52, 0x08, 0x64, 0x69, 0x73, 0x61, 0x62, 0x6c, 0x65, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x76, 0x65,
+	0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x07, 0x76, 0x65, 0x72,
+	0x73, 0x69, 0x6f, 0x6e, 0x42, 0x3a, 0x5a, 0x38, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63,
+	0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x77, 0x61, 0x74,
+	0x63, 0x68, 0x74, 0x6f, 0x77, 0x65, 0x72, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c,
+	0x2f, 0x69, 0x61, 0x6d, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65,
+	0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -205,6 +220,7 @@ func file_controller_storage_iam_store_v1_group_proto_init() {
 	if File_controller_storage_iam_store_v1_group_proto != nil {
 		return
 	}
+	file_controller_storage_iam_store_v1_scope_proto_init()
 	if !protoimpl.UnsafeEnabled {
 		file_controller_storage_iam_store_v1_group_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*Group); i {
diff --git a/internal/iam/store/group_member.pb.go b/internal/iam/store/group_member.pb.go
new file mode 100644
index 000000000..fee07372b
--- /dev/null
+++ b/internal/iam/store/group_member.pb.go
@@ -0,0 +1,333 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.24.0
+// 	protoc        v3.12.3
+// source: controller/storage/iam/store/v1/group_member.proto
+
+package store
+
+import (
+	proto "github.com/golang/protobuf/proto"
+	timestamp "github.com/hashicorp/watchtower/internal/db/timestamp"
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+// This is a compile-time assertion that a sufficiently up-to-date version
+// of the legacy proto package is being used.
+const _ = proto.ProtoPackageIsVersion4
+
+type GroupMemberUser struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// create_time from the RDBMS
+	// @inject_tag: `gorm:"default:current_timestamp"`
+	CreateTime *timestamp.Timestamp `protobuf:"bytes,1,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty" gorm:"default:current_timestamp"`
+	// group_id is the group of this member.
+	// @inject_tag: gorm:"primary_key"
+	GroupId string `protobuf:"bytes,2,opt,name=group_id,json=groupId,proto3" json:"group_id,omitempty" gorm:"primary_key"`
+	// member_id is the public_id of the user (which is the member)
+	// @inject_tag: gorm:"primary_key"
+	MemberId string `protobuf:"bytes,3,opt,name=member_id,json=memberId,proto3" json:"member_id,omitempty" gorm:"primary_key"`
+}
+
+func (x *GroupMemberUser) Reset() {
+	*x = GroupMemberUser{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_controller_storage_iam_store_v1_group_member_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *GroupMemberUser) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*GroupMemberUser) ProtoMessage() {}
+
+func (x *GroupMemberUser) ProtoReflect() protoreflect.Message {
+	mi := &file_controller_storage_iam_store_v1_group_member_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use GroupMemberUser.ProtoReflect.Descriptor instead.
+func (*GroupMemberUser) Descriptor() ([]byte, []int) {
+	return file_controller_storage_iam_store_v1_group_member_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *GroupMemberUser) GetCreateTime() *timestamp.Timestamp {
+	if x != nil {
+		return x.CreateTime
+	}
+	return nil
+}
+
+func (x *GroupMemberUser) GetGroupId() string {
+	if x != nil {
+		return x.GroupId
+	}
+	return ""
+}
+
+func (x *GroupMemberUser) GetMemberId() string {
+	if x != nil {
+		return x.MemberId
+	}
+	return ""
+}
+
+type GroupMemberView struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// create_time from the RDBMS
+	// @inject_tag: `gorm:"default:current_timestamp"`
+	CreateTime *timestamp.Timestamp `protobuf:"bytes,1,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty" gorm:"default:current_timestamp"`
+	// group_id is the group of this member.
+	// @inject_tag: gorm:"primary_key"
+	GroupId string `protobuf:"bytes,2,opt,name=group_id,json=groupId,proto3" json:"group_id,omitempty" gorm:"primary_key"`
+	// member_id is the public_id of member
+	// @inject_tag: gorm:"primary_key"
+	MemberId string `protobuf:"bytes,3,opt,name=member_id,json=memberId,proto3" json:"member_id,omitempty" gorm:"primary_key"`
+	// Principal type (User or Group)
+	Type string `protobuf:"bytes,4,opt,name=type,proto3" json:"type,omitempty"`
+	// member_scope_id of the member
+	// @inject_tag: `gorm:"default:null"`
+	MemberScopeId string `protobuf:"bytes,5,opt,name=member_scope_id,json=memberScopeId,proto3" json:"member_scope_id,omitempty" gorm:"default:null"`
+	// group_scope_id of the role
+	// @inject_tag: `gorm:"default:null"`
+	GroupScopeId string `protobuf:"bytes,6,opt,name=group_scope_id,json=groupScopeId,proto3" json:"group_scope_id,omitempty" gorm:"default:null"`
+	// scoped_member_id of the member
+	// @inject_tag: `gorm:"default:null"`
+	ScopedMemberId string `protobuf:"bytes,7,opt,name=scoped_member_id,json=scopedMemberId,proto3" json:"scoped_member_id,omitempty" gorm:"default:null"`
+}
+
+func (x *GroupMemberView) Reset() {
+	*x = GroupMemberView{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_controller_storage_iam_store_v1_group_member_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *GroupMemberView) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*GroupMemberView) ProtoMessage() {}
+
+func (x *GroupMemberView) ProtoReflect() protoreflect.Message {
+	mi := &file_controller_storage_iam_store_v1_group_member_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use GroupMemberView.ProtoReflect.Descriptor instead.
+func (*GroupMemberView) Descriptor() ([]byte, []int) {
+	return file_controller_storage_iam_store_v1_group_member_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *GroupMemberView) GetCreateTime() *timestamp.Timestamp {
+	if x != nil {
+		return x.CreateTime
+	}
+	return nil
+}
+
+func (x *GroupMemberView) GetGroupId() string {
+	if x != nil {
+		return x.GroupId
+	}
+	return ""
+}
+
+func (x *GroupMemberView) GetMemberId() string {
+	if x != nil {
+		return x.MemberId
+	}
+	return ""
+}
+
+func (x *GroupMemberView) GetType() string {
+	if x != nil {
+		return x.Type
+	}
+	return ""
+}
+
+func (x *GroupMemberView) GetMemberScopeId() string {
+	if x != nil {
+		return x.MemberScopeId
+	}
+	return ""
+}
+
+func (x *GroupMemberView) GetGroupScopeId() string {
+	if x != nil {
+		return x.GroupScopeId
+	}
+	return ""
+}
+
+func (x *GroupMemberView) GetScopedMemberId() string {
+	if x != nil {
+		return x.ScopedMemberId
+	}
+	return ""
+}
+
+var File_controller_storage_iam_store_v1_group_member_proto protoreflect.FileDescriptor
+
+var file_controller_storage_iam_store_v1_group_member_proto_rawDesc = []byte{
+	0x0a, 0x32, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f,
+	0x72, 0x61, 0x67, 0x65, 0x2f, 0x69, 0x61, 0x6d, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x2f, 0x76,
+	0x31, 0x2f, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x5f, 0x6d, 0x65, 0x6d, 0x62, 0x65, 0x72, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x12, 0x1f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72,
+	0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x69, 0x61, 0x6d, 0x2e, 0x73, 0x74, 0x6f,
+	0x72, 0x65, 0x2e, 0x76, 0x31, 0x1a, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65,
+	0x72, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74,
+	0x61, 0x6d, 0x70, 0x2f, 0x76, 0x31, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70,
+	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x2b, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c,
+	0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2f, 0x69, 0x61, 0x6d, 0x2f, 0x73,
+	0x74, 0x6f, 0x72, 0x65, 0x2f, 0x76, 0x31, 0x2f, 0x73, 0x63, 0x6f, 0x70, 0x65, 0x2e, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x22, 0x96, 0x01, 0x0a, 0x0f, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x4d, 0x65, 0x6d,
+	0x62, 0x65, 0x72, 0x55, 0x73, 0x65, 0x72, 0x12, 0x4b, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61, 0x74,
+	0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63,
+	0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67,
+	0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54,
+	0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65,
+	0x54, 0x69, 0x6d, 0x65, 0x12, 0x19, 0x0a, 0x08, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x5f, 0x69, 0x64,
+	0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x49, 0x64, 0x12,
+	0x1b, 0x0a, 0x09, 0x6d, 0x65, 0x6d, 0x62, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01,
+	0x28, 0x09, 0x52, 0x08, 0x6d, 0x65, 0x6d, 0x62, 0x65, 0x72, 0x49, 0x64, 0x22, 0xa2, 0x02, 0x0a,
+	0x0f, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x4d, 0x65, 0x6d, 0x62, 0x65, 0x72, 0x56, 0x69, 0x65, 0x77,
+	0x12, 0x4b, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18,
+	0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c,
+	0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73,
+	0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d,
+	0x70, 0x52, 0x0a, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x19, 0x0a,
+	0x08, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x07, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x49, 0x64, 0x12, 0x1b, 0x0a, 0x09, 0x6d, 0x65, 0x6d, 0x62,
+	0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x6d, 0x65, 0x6d,
+	0x62, 0x65, 0x72, 0x49, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x04, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x26, 0x0a, 0x0f, 0x6d, 0x65, 0x6d,
+	0x62, 0x65, 0x72, 0x5f, 0x73, 0x63, 0x6f, 0x70, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x05, 0x20, 0x01,
+	0x28, 0x09, 0x52, 0x0d, 0x6d, 0x65, 0x6d, 0x62, 0x65, 0x72, 0x53, 0x63, 0x6f, 0x70, 0x65, 0x49,
+	0x64, 0x12, 0x24, 0x0a, 0x0e, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x5f, 0x73, 0x63, 0x6f, 0x70, 0x65,
+	0x5f, 0x69, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x67, 0x72, 0x6f, 0x75, 0x70,
+	0x53, 0x63, 0x6f, 0x70, 0x65, 0x49, 0x64, 0x12, 0x28, 0x0a, 0x10, 0x73, 0x63, 0x6f, 0x70, 0x65,
+	0x64, 0x5f, 0x6d, 0x65, 0x6d, 0x62, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x07, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x0e, 0x73, 0x63, 0x6f, 0x70, 0x65, 0x64, 0x4d, 0x65, 0x6d, 0x62, 0x65, 0x72, 0x49,
+	0x64, 0x42, 0x3a, 0x5a, 0x38, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f,
+	0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x77, 0x61, 0x74, 0x63, 0x68, 0x74,
+	0x6f, 0x77, 0x65, 0x72, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x69, 0x61,
+	0x6d, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62, 0x06, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_controller_storage_iam_store_v1_group_member_proto_rawDescOnce sync.Once
+	file_controller_storage_iam_store_v1_group_member_proto_rawDescData = file_controller_storage_iam_store_v1_group_member_proto_rawDesc
+)
+
+func file_controller_storage_iam_store_v1_group_member_proto_rawDescGZIP() []byte {
+	file_controller_storage_iam_store_v1_group_member_proto_rawDescOnce.Do(func() {
+		file_controller_storage_iam_store_v1_group_member_proto_rawDescData = protoimpl.X.CompressGZIP(file_controller_storage_iam_store_v1_group_member_proto_rawDescData)
+	})
+	return file_controller_storage_iam_store_v1_group_member_proto_rawDescData
+}
+
+var file_controller_storage_iam_store_v1_group_member_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
+var file_controller_storage_iam_store_v1_group_member_proto_goTypes = []interface{}{
+	(*GroupMemberUser)(nil),     // 0: controller.storage.iam.store.v1.GroupMemberUser
+	(*GroupMemberView)(nil),     // 1: controller.storage.iam.store.v1.GroupMemberView
+	(*timestamp.Timestamp)(nil), // 2: controller.storage.timestamp.v1.Timestamp
+}
+var file_controller_storage_iam_store_v1_group_member_proto_depIdxs = []int32{
+	2, // 0: controller.storage.iam.store.v1.GroupMemberUser.create_time:type_name -> controller.storage.timestamp.v1.Timestamp
+	2, // 1: controller.storage.iam.store.v1.GroupMemberView.create_time:type_name -> controller.storage.timestamp.v1.Timestamp
+	2, // [2:2] is the sub-list for method output_type
+	2, // [2:2] is the sub-list for method input_type
+	2, // [2:2] is the sub-list for extension type_name
+	2, // [2:2] is the sub-list for extension extendee
+	0, // [0:2] is the sub-list for field type_name
+}
+
+func init() { file_controller_storage_iam_store_v1_group_member_proto_init() }
+func file_controller_storage_iam_store_v1_group_member_proto_init() {
+	if File_controller_storage_iam_store_v1_group_member_proto != nil {
+		return
+	}
+	file_controller_storage_iam_store_v1_scope_proto_init()
+	if !protoimpl.UnsafeEnabled {
+		file_controller_storage_iam_store_v1_group_member_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*GroupMemberUser); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_controller_storage_iam_store_v1_group_member_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*GroupMemberView); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_controller_storage_iam_store_v1_group_member_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   2,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_controller_storage_iam_store_v1_group_member_proto_goTypes,
+		DependencyIndexes: file_controller_storage_iam_store_v1_group_member_proto_depIdxs,
+		MessageInfos:      file_controller_storage_iam_store_v1_group_member_proto_msgTypes,
+	}.Build()
+	File_controller_storage_iam_store_v1_group_member_proto = out.File
+	file_controller_storage_iam_store_v1_group_member_proto_rawDesc = nil
+	file_controller_storage_iam_store_v1_group_member_proto_goTypes = nil
+	file_controller_storage_iam_store_v1_group_member_proto_depIdxs = nil
+}
diff --git a/internal/iam/testing.go b/internal/iam/testing.go
index 4077d107c..03549f7cb 100644
--- a/internal/iam/testing.go
+++ b/internal/iam/testing.go
@@ -154,6 +154,19 @@ func TestGroup(t *testing.T, conn *gorm.DB, scopeId string, opt ...Option) *Grou
 	return grp
 }
 
+func TestGroupMember(t *testing.T, conn *gorm.DB, groupId, userId string, opt ...Option) *GroupMemberUser {
+	t.Helper()
+	require := require.New(t)
+	rw := db.New(conn)
+	gm, err := NewGroupMemberUser(groupId, userId)
+	require.NoError(err)
+	require.NotNil(gm)
+	err = rw.Create(context.Background(), gm)
+	require.NoError(err)
+	require.NotEmpty(gm.CreateTime)
+	return gm
+}
+
 func TestUserRole(t *testing.T, conn *gorm.DB, roleId, userId string, opt ...Option) *UserRole {
 	t.Helper()
 	require := require.New(t)
diff --git a/internal/iam/testing_test.go b/internal/iam/testing_test.go
index 057a2270d..3cec2e879 100644
--- a/internal/iam/testing_test.go
+++ b/internal/iam/testing_test.go
@@ -169,3 +169,24 @@ func Test_TestGroupRole(t *testing.T) {
 	require.Equal(projRole.PublicId, groupRole.RoleId)
 	require.Equal(proj2Group.PublicId, groupRole.PrincipalId)
 }
+
+func Test_TestGroupMember(t *testing.T) {
+	t.Helper()
+	require := require.New(t)
+	conn, _ := db.TestSetup(t, "postgres")
+	org, proj := TestScopes(t, conn)
+	og := TestGroup(t, conn, org.PublicId)
+	pg := TestGroup(t, conn, proj.PublicId)
+	u := TestUser(t, conn, org.PublicId)
+
+	gm := TestGroupMember(t, conn, og.PublicId, u.PublicId)
+	require.NotNil(gm)
+	require.Equal(og.PublicId, gm.GroupId)
+	require.Equal(u.PublicId, gm.MemberId)
+
+	gm = TestGroupMember(t, conn, pg.PublicId, u.PublicId)
+	require.NotNil(gm)
+	require.Equal(pg.PublicId, gm.GroupId)
+	require.Equal(u.PublicId, gm.MemberId)
+
+}
diff --git a/internal/proto/local/controller/storage/iam/store/v1/group.proto b/internal/proto/local/controller/storage/iam/store/v1/group.proto
index 69f813ec2..9b3ecdc68 100644
--- a/internal/proto/local/controller/storage/iam/store/v1/group.proto
+++ b/internal/proto/local/controller/storage/iam/store/v1/group.proto
@@ -4,6 +4,7 @@ package controller.storage.iam.store.v1;
 option go_package = "github.com/hashicorp/watchtower/internal/iam/store;store";
 
 import "controller/storage/timestamp/v1/timestamp.proto";
+import "controller/storage/iam/store/v1/scope.proto";
 
 message Group {
   // public_id is used to access the Group via an API
@@ -34,4 +35,9 @@ message Group {
   // disabled is by default false and allows a Group to be marked disabled.
   // @inject_tag: `gorm:"default:null"`
   bool disabled = 7;
+
+  // version allows optimistic locking of the group when modifying the group
+  // itself and when modifying dependent items like group members.
+  // @inject_tag: `gorm:"default:null"`
+  uint32 version = 8;
 }
\ No newline at end of file
diff --git a/internal/proto/local/controller/storage/iam/store/v1/group_member.proto b/internal/proto/local/controller/storage/iam/store/v1/group_member.proto
new file mode 100644
index 000000000..186588e6d
--- /dev/null
+++ b/internal/proto/local/controller/storage/iam/store/v1/group_member.proto
@@ -0,0 +1,50 @@
+syntax = "proto3";
+
+package controller.storage.iam.store.v1;
+option go_package = "github.com/hashicorp/watchtower/internal/iam/store;store";
+
+import "controller/storage/timestamp/v1/timestamp.proto";
+import "controller/storage/iam/store/v1/scope.proto";
+
+message GroupMemberUser {
+  // create_time from the RDBMS
+  // @inject_tag: `gorm:"default:current_timestamp"`
+  timestamp.v1.Timestamp create_time = 1;
+
+  // group_id is the group of this member.
+  // @inject_tag: gorm:"primary_key"
+  string group_id = 2;
+
+  // member_id is the public_id of the user (which is the member)
+  // @inject_tag: gorm:"primary_key"
+  string member_id = 3;
+}
+
+message GroupMemberView {
+  // create_time from the RDBMS
+  // @inject_tag: `gorm:"default:current_timestamp"`
+  timestamp.v1.Timestamp create_time = 1;
+
+  // group_id is the group of this member.
+  // @inject_tag: gorm:"primary_key"
+  string group_id = 2;
+
+  // member_id is the public_id of member
+  // @inject_tag: gorm:"primary_key"
+  string member_id = 3;
+
+  // Principal type (User or Group)
+  string type = 4;
+
+  // member_scope_id of the member
+  // @inject_tag: `gorm:"default:null"`
+  string member_scope_id = 5;
+
+  // group_scope_id of the role
+  // @inject_tag: `gorm:"default:null"`
+  string group_scope_id = 6;
+
+  // scoped_member_id of the member
+  // @inject_tag: `gorm:"default:null"`
+  string scoped_member_id = 7;
+}