configstore: implement organization members

internal/services/configstore/api/client.go

@@ -390,6 +390,12 @@ func (c *Client) DeleteUserToken(ctx context.Context, userRef, tokenName string)
 	return c.getResponse(ctx, "DELETE", fmt.Sprintf("/users/%s/tokens/%s", userRef, tokenName), nil, jsonContent, nil)
 }
 
+func (c *Client) GetUserOrgs(ctx context.Context, userRef string) ([]*UserOrgsResponse, *http.Response, error) {
+	userOrgs := []*UserOrgsResponse{}
+	resp, err := c.getParsedResponse(ctx, "GET", fmt.Sprintf("/users/%s/orgs", userRef), nil, jsonContent, nil, &userOrgs)
+	return userOrgs, resp, err
+}
+
 func (c *Client) GetRemoteSource(ctx context.Context, rsRef string) (*types.RemoteSource, *http.Response, error) {
 	rs := new(types.RemoteSource)
 	resp, err := c.getParsedResponse(ctx, "GET", fmt.Sprintf("/remotesources/%s", rsRef), nil, jsonContent, nil, rs)

internal/services/configstore/api/user.go

@@ -496,8 +496,51 @@ func (h *DeleteUserTokenHandler) ServeHTTP(w http.ResponseWriter, r *http.Reques
 	err := h.ch.DeleteUserToken(ctx, userRef, tokenName)
 	if httpError(w, err) {
 		h.log.Errorf("err: %+v", err)
+		return
 	}
 	if err := httpResponse(w, http.StatusNoContent, nil); err != nil {
 		h.log.Errorf("err: %+v", err)
 	}
 }
+
+type UserOrgsResponse struct {
+	Organization *types.Organization
+	Role         types.MemberRole
+}
+
+func userOrgsResponse(userOrg *command.UserOrgsResponse) *UserOrgsResponse {
+	return &UserOrgsResponse{
+		Organization: userOrg.Organization,
+		Role:         userOrg.Role,
+	}
+}
+
+type UserOrgsHandler struct {
+	log *zap.SugaredLogger
+	ch  *command.CommandHandler
+}
+
+func NewUserOrgsHandler(logger *zap.Logger, ch *command.CommandHandler) *UserOrgsHandler {
+	return &UserOrgsHandler{log: logger.Sugar(), ch: ch}
+}
+
+func (h *UserOrgsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	vars := mux.Vars(r)
+	userRef := vars["userref"]
+
+	userOrgs, err := h.ch.GetUserOrgs(ctx, userRef)
+	if httpError(w, err) {
+		h.log.Errorf("err: %+v", err)
+		return
+	}
+
+	res := make([]*UserOrgsResponse, len(userOrgs))
+	for i, userOrg := range userOrgs {
+		res[i] = userOrgsResponse(userOrg)
+	}
+
+	if err := httpResponse(w, http.StatusOK, res); err != nil {
+		h.log.Errorf("err: %+v", err)
+	}
+}

internal/services/configstore/command/org.go

@@ -73,12 +73,40 @@ func (s *CommandHandler) CreateOrg(ctx context.Context, org *types.Organization)
 		return nil, err
 	}
 
+	actions := []*datamanager.Action{}
+
 	org.ID = uuid.NewV4().String()
 	org.CreatedAt = time.Now()
 	orgj, err := json.Marshal(org)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to marshal org")
 	}
+	actions = append(actions, &datamanager.Action{
+		ActionType: datamanager.ActionTypePut,
+		DataType:   string(types.ConfigTypeOrg),
+		ID:         org.ID,
+		Data:       orgj,
+	})
+
+	if org.CreatorUserID != "" {
+		// add the creator as org member with role owner
+		orgmember := &types.OrganizationMember{
+			ID:             uuid.NewV4().String(),
+			OrganizationID: org.ID,
+			UserID:         org.CreatorUserID,
+			MemberRole:     types.MemberRoleOwner,
+		}
+		orgmemberj, err := json.Marshal(orgmember)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to marshal project group")
+		}
+		actions = append(actions, &datamanager.Action{
+			ActionType: datamanager.ActionTypePut,
+			DataType:   string(types.ConfigTypeOrgMember),
+			ID:         orgmember.ID,
+			Data:       orgmemberj,
+		})
+	}
 
 	pg := &types.ProjectGroup{
 		ID: uuid.NewV4().String(),
@@ -91,20 +119,12 @@ func (s *CommandHandler) CreateOrg(ctx context.Context, org *types.Organization)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to marshal project group")
 	}
-	actions := []*datamanager.Action{
+	actions = append(actions, &datamanager.Action{
-		{
+		ActionType: datamanager.ActionTypePut,
-			ActionType: datamanager.ActionTypePut,
+		DataType:   string(types.ConfigTypeProjectGroup),
-			DataType:   string(types.ConfigTypeOrg),
+		ID:         pg.ID,
-			ID:         org.ID,
+		Data:       pgj,
-			Data:       orgj,
+	})
-		},
-		{
-			ActionType: datamanager.ActionTypePut,
-			DataType:   string(types.ConfigTypeProjectGroup),
-			ID:         pg.ID,
-			Data:       pgj,
-		},
-	}
 
 	_, err = s.dm.WriteWal(ctx, actions, cgt)
 	return org, err
@@ -115,7 +135,6 @@ func (s *CommandHandler) DeleteOrg(ctx context.Context, orgRef string) error {
 	var projects []*types.Project
 
 	var cgt *datamanager.ChangeGroupsUpdateToken
-
 	// must do all the checks in a single transaction to avoid concurrent changes
 	err := s.readDB.Do(func(tx *db.Tx) error {
 		var err error

internal/services/configstore/command/user.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/sorintlab/agola/internal/datamanager"
 	"github.com/sorintlab/agola/internal/db"
+	"github.com/sorintlab/agola/internal/services/configstore/readdb"
 	"github.com/sorintlab/agola/internal/services/types"
 	"github.com/sorintlab/agola/internal/util"
 
@@ -624,3 +625,42 @@ func (s *CommandHandler) DeleteUserToken(ctx context.Context, userRef, tokenName
 	_, err = s.dm.WriteWal(ctx, actions, cgt)
 	return err
 }
+
+type UserOrgsResponse struct {
+	Organization *types.Organization
+	Role         types.MemberRole
+}
+
+func userOrgsResponse(userOrg *readdb.UserOrg) *UserOrgsResponse {
+	return &UserOrgsResponse{
+		Organization: userOrg.Organization,
+		Role:         userOrg.Role,
+	}
+}
+
+func (s *CommandHandler) GetUserOrgs(ctx context.Context, userRef string) ([]*UserOrgsResponse, error) {
+	var userOrgs []*readdb.UserOrg
+	err := s.readDB.Do(func(tx *db.Tx) error {
+		var err error
+		user, err := s.readDB.GetUser(tx, userRef)
+		if err != nil {
+			return err
+		}
+		if user == nil {
+			return util.NewErrNotFound(errors.Errorf("user %q doesn't exist", userRef))
+		}
+
+		userOrgs, err = s.readDB.GetUserOrgs(tx, user.ID)
+		return err
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	res := make([]*UserOrgsResponse, len(userOrgs))
+	for i, userOrg := range userOrgs {
+		res[i] = userOrgsResponse(userOrg)
+	}
+
+	return res, nil
+}

internal/services/configstore/configstore.go

@@ -78,6 +78,7 @@ func NewConfigStore(ctx context.Context, c *config.ConfigStore) (*ConfigStore, e
 		DataTypes: []string{
 			string(types.ConfigTypeUser),
 			string(types.ConfigTypeOrg),
+			string(types.ConfigTypeOrgMember),
 			string(types.ConfigTypeProjectGroup),
 			string(types.ConfigTypeProject),
 			string(types.ConfigTypeRemoteSource),
@@ -154,6 +155,8 @@ func (s *ConfigStore) Run(ctx context.Context) error {
 	createUserTokenHandler := api.NewCreateUserTokenHandler(logger, s.ch)
 	deleteUserTokenHandler := api.NewDeleteUserTokenHandler(logger, s.ch)
 
+	userOrgsHandler := api.NewUserOrgsHandler(logger, s.ch)
+
 	orgHandler := api.NewOrgHandler(logger, s.readDB)
 	orgsHandler := api.NewOrgsHandler(logger, s.readDB)
 	createOrgHandler := api.NewCreateOrgHandler(logger, s.ch)
@@ -202,6 +205,8 @@ func (s *ConfigStore) Run(ctx context.Context) error {
 	apirouter.Handle("/users/{userref}/tokens", createUserTokenHandler).Methods("POST")
 	apirouter.Handle("/users/{userref}/tokens/{tokenname}", deleteUserTokenHandler).Methods("DELETE")
 
+	apirouter.Handle("/users/{userref}/orgs", userOrgsHandler).Methods("GET")
+
 	apirouter.Handle("/orgs/{orgref}", orgHandler).Methods("GET")
 	apirouter.Handle("/orgs", orgsHandler).Methods("GET")
 	apirouter.Handle("/orgs", createOrgHandler).Methods("POST")

internal/services/configstore/configstore_test.go

@@ -26,6 +26,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/google/go-cmp/cmp"
 	"github.com/sorintlab/agola/internal/db"
 	"github.com/sorintlab/agola/internal/services/config"
 	"github.com/sorintlab/agola/internal/services/configstore/command"
@@ -505,3 +506,97 @@ func TestProjectGroupsAndProjects(t *testing.T) {
 		}
 	})
 }
+
+func TestOrgMembers(t *testing.T) {
+	dir, err := ioutil.TempDir("", "agola")
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+	defer os.RemoveAll(dir)
+
+	ctx := context.Background()
+
+	cs, tetcd := setupConfigstore(t, ctx, dir)
+	defer shutdownEtcd(tetcd)
+
+	t.Logf("starting cs")
+	go func() {
+		if err := cs.Run(ctx); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	}()
+
+	// TODO(sgotti) change the sleep with a real check that all is ready
+	time.Sleep(2 * time.Second)
+
+	user, err := cs.ch.CreateUser(ctx, &command.CreateUserRequest{UserName: "user01"})
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+	org, err := cs.ch.CreateOrg(ctx, &types.Organization{Name: "org01", CreatorUserID: user.ID})
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+
+	// TODO(sgotti) change the sleep with a real check that all is ready
+	time.Sleep(2 * time.Second)
+
+	t.Run("test user org creator is org member with owner role", func(t *testing.T) {
+		expectedResponse := []*command.UserOrgsResponse{
+			{
+				Organization: org,
+				Role:         types.MemberRoleOwner,
+			},
+		}
+		res, err := cs.ch.GetUserOrgs(ctx, user.ID)
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+		if diff := cmp.Diff(res, expectedResponse); diff != "" {
+			t.Error(diff)
+		}
+	})
+
+	orgs := []*types.Organization{}
+	for i := 0; i < 10; i++ {
+		org, err := cs.ch.CreateOrg(ctx, &types.Organization{Name: fmt.Sprintf("org%d", i), CreatorUserID: user.ID})
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+		orgs = append(orgs, org)
+		time.Sleep(200 * time.Millisecond)
+	}
+
+	for i := 0; i < 5; i++ {
+		if err := cs.ch.DeleteOrg(ctx, fmt.Sprintf("org%d", i)); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	}
+
+	// delete some org and check that if also orgmembers aren't yet cleaned only the existing orgs are reported
+	t.Run("test only existing orgs are reported", func(t *testing.T) {
+		expectedResponse := []*command.UserOrgsResponse{
+			{
+				Organization: org,
+				Role:         types.MemberRoleOwner,
+			},
+		}
+		for i := 5; i < 10; i++ {
+			expectedResponse = append(expectedResponse, &command.UserOrgsResponse{
+				Organization: orgs[i],
+				Role:         types.MemberRoleOwner,
+			})
+		}
+		res, err := cs.ch.GetUserOrgs(ctx, user.ID)
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+		if diff := cmp.Diff(res, expectedResponse); diff != "" {
+			t.Error(diff)
+		}
+	})
+
+	// TODO(sgotti) change the sleep with a real check that user is in readdb
+	time.Sleep(2 * time.Second)
+
+}

internal/services/configstore/readdb/create.go

@@ -38,6 +38,10 @@ var Stmts = []string{
 	"create table org (id uuid, name varchar, data bytea, PRIMARY KEY (id))",
 	"create index org_name on org(name)",
 
+	"create table orgmember (id uuid, orgid uuid, userid uuid, role varchar, data bytea, PRIMARY KEY (id))",
+	"create index orgmember_role on orgmember(role)",
+	"create index orgmember_orgid_userid on orgmember(orgid, userid)",
+
 	"create table remotesource (id uuid, name varchar, data bytea, PRIMARY KEY (id))",
 
 	"create table linkedaccount_user (id uuid, remotesourceid uuid, userid uuid, remoteuserid uuid, PRIMARY KEY (id), FOREIGN KEY(userid) REFERENCES user(id))",

internal/services/configstore/readdb/org.go

@@ -30,6 +30,9 @@ import (
 var (
 	orgSelect = sb.Select("org.id", "org.data").From("org")
 	orgInsert = sb.Insert("org").Columns("id", "name", "data")
+
+	orgmemberSelect = sb.Select("orgmember.id", "orgmember.data").From("orgmember")
+	orgmemberInsert = sb.Insert("orgmember").Columns("id", "orgid", "userid", "role", "data")
 )
 
 func (r *ReadDB) insertOrg(tx *db.Tx, data []byte) error {
@@ -187,12 +190,12 @@ func scanOrgs(rows *sql.Rows) ([]*types.Organization, []string, error) {
 	orgs := []*types.Organization{}
 	ids := []string{}
 	for rows.Next() {
-		p, id, err := scanOrg(rows)
+		org, id, err := scanOrg(rows)
 		if err != nil {
 			rows.Close()
 			return nil, nil, err
 		}
-		orgs = append(orgs, p)
+		orgs = append(orgs, org)
 		ids = append(ids, id)
 	}
 	if err := rows.Err(); err != nil {
@@ -200,3 +203,176 @@ func scanOrgs(rows *sql.Rows) ([]*types.Organization, []string, error) {
 	}
 	return orgs, ids, nil
 }
+
+func (r *ReadDB) insertOrgMember(tx *db.Tx, data []byte) error {
+	orgmember := types.OrganizationMember{}
+	if err := json.Unmarshal(data, &orgmember); err != nil {
+		return errors.Wrap(err, "failed to unmarshal orgmember")
+	}
+	r.log.Infof("inserting orgmember: %s", util.Dump(orgmember))
+	// poor man insert or update...
+	if err := r.deleteOrgMember(tx, orgmember.ID); err != nil {
+		return err
+	}
+	q, args, err := orgmemberInsert.Values(orgmember.ID, orgmember.OrganizationID, orgmember.UserID, orgmember.MemberRole, data).ToSql()
+	if err != nil {
+		return errors.Wrap(err, "failed to build query")
+	}
+	if _, err := tx.Exec(q, args...); err != nil {
+		return errors.Wrap(err, "failed to insert orgmember")
+	}
+
+	return nil
+}
+
+func (r *ReadDB) deleteOrgMember(tx *db.Tx, orgmemberID string) error {
+	if _, err := tx.Exec("delete from orgmember where id = $1", orgmemberID); err != nil {
+		return errors.Wrap(err, "failed to delete orgmember")
+	}
+	return nil
+}
+
+func fetchOrgMembers(tx *db.Tx, q string, args ...interface{}) ([]*types.OrganizationMember, []string, error) {
+	rows, err := tx.Query(q, args...)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer rows.Close()
+	return scanOrgMembers(rows)
+}
+
+func scanOrgMember(rows *sql.Rows, additionalFields ...interface{}) (*types.OrganizationMember, string, error) {
+	var id string
+	var data []byte
+	if err := rows.Scan(&id, &data); err != nil {
+		return nil, "", errors.Wrap(err, "failed to scan rows")
+	}
+	orgmember := types.OrganizationMember{}
+	if len(data) > 0 {
+		if err := json.Unmarshal(data, &orgmember); err != nil {
+			return nil, "", errors.Wrap(err, "failed to unmarshal org")
+		}
+	}
+
+	return &orgmember, id, nil
+}
+
+func scanOrgMembers(rows *sql.Rows) ([]*types.OrganizationMember, []string, error) {
+	orgmembers := []*types.OrganizationMember{}
+	ids := []string{}
+	for rows.Next() {
+		orgmember, id, err := scanOrgMember(rows)
+		if err != nil {
+			rows.Close()
+			return nil, nil, err
+		}
+		orgmembers = append(orgmembers, orgmember)
+		ids = append(ids, id)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, nil, err
+	}
+	return orgmembers, ids, nil
+}
+
+type OrgUser struct {
+	User *types.User
+	Role types.MemberRole
+}
+
+// TODO(sgotti) implement cursor fetching
+func (r *ReadDB) GetOrgUsers(tx *db.Tx, orgID string) ([]*OrgUser, error) {
+	s := sb.Select("orgmember.data", "user.data").From("orgmember")
+	s = s.Where(sq.Eq{"orgmember.orgid": orgID})
+	s = s.Join("user on user.id = orgmember.userid")
+	s = s.OrderBy("user.name")
+	q, args, err := s.ToSql()
+	r.log.Debugf("q: %s, args: %s", q, util.Dump(args))
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to build query")
+	}
+
+	rows, err := tx.Query(q, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	orgusers := []*OrgUser{}
+	for rows.Next() {
+		var orgmember *types.OrganizationMember
+		var user *types.User
+		var orgmemberdata []byte
+		var userdata []byte
+		if err := rows.Scan(&orgmemberdata, &userdata); err != nil {
+			return nil, errors.Wrap(err, "failed to scan rows")
+		}
+		if err := json.Unmarshal(orgmemberdata, &orgmember); err != nil {
+			return nil, errors.Wrap(err, "failed to unmarshal orgmember")
+		}
+		if err := json.Unmarshal(userdata, &user); err != nil {
+			return nil, errors.Wrap(err, "failed to unmarshal org")
+		}
+
+		orgusers = append(orgusers, &OrgUser{
+			User: user,
+			Role: orgmember.MemberRole,
+		})
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	return orgusers, nil
+}
+
+type UserOrg struct {
+	Organization *types.Organization
+	Role         types.MemberRole
+}
+
+// TODO(sgotti) implement cursor fetching
+func (r *ReadDB) GetUserOrgs(tx *db.Tx, userID string) ([]*UserOrg, error) {
+	s := sb.Select("orgmember.data", "org.data").From("orgmember")
+	s = s.Where(sq.Eq{"orgmember.userid": userID})
+	s = s.Join("org on org.id = orgmember.orgid")
+	s = s.OrderBy("org.name")
+	q, args, err := s.ToSql()
+	r.log.Debugf("q: %s, args: %s", q, util.Dump(args))
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to build query")
+	}
+
+	rows, err := tx.Query(q, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	userorgs := []*UserOrg{}
+	for rows.Next() {
+		var orgmember *types.OrganizationMember
+		var org *types.Organization
+		var orgmemberdata []byte
+		var orgdata []byte
+		if err := rows.Scan(&orgmemberdata, &orgdata); err != nil {
+			return nil, errors.Wrap(err, "failed to scan rows")
+		}
+		if err := json.Unmarshal(orgmemberdata, &orgmember); err != nil {
+			return nil, errors.Wrap(err, "failed to unmarshal orgmember")
+		}
+		if err := json.Unmarshal(orgdata, &org); err != nil {
+			return nil, errors.Wrap(err, "failed to unmarshal org")
+		}
+
+		userorgs = append(userorgs, &UserOrg{
+			Organization: org,
+			Role:         orgmember.MemberRole,
+		})
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	return userorgs, nil
+}

internal/services/configstore/readdb/readdb.go

@@ -587,6 +587,10 @@ func (r *ReadDB) applyAction(tx *db.Tx, action *datamanager.Action) error {
 			if err := r.insertOrg(tx, action.Data); err != nil {
 				return err
 			}
+		case types.ConfigTypeOrgMember:
+			if err := r.insertOrgMember(tx, action.Data); err != nil {
+				return err
+			}
 		case types.ConfigTypeProjectGroup:
 			if err := r.insertProjectGroup(tx, action.Data); err != nil {
 				return err
@@ -621,6 +625,11 @@ func (r *ReadDB) applyAction(tx *db.Tx, action *datamanager.Action) error {
 			if err := r.deleteOrg(tx, action.ID); err != nil {
 				return err
 			}
+		case types.ConfigTypeOrgMember:
+			r.log.Debugf("deleting orgmember with id: %s", action.ID)
+			if err := r.deleteOrgMember(tx, action.ID); err != nil {
+				return err
+			}
 		case types.ConfigTypeProjectGroup:
 			r.log.Debugf("deleting project group with id: %s", action.ID)
 			if err := r.deleteProjectGroup(tx, action.ID); err != nil {

internal/services/types/types.go

@@ -27,6 +27,7 @@ type ConfigType string
 const (
 	ConfigTypeUser         ConfigType = "user"
 	ConfigTypeOrg          ConfigType = "org"
+	ConfigTypeOrgMember    ConfigType = "orgmember"
 	ConfigTypeProjectGroup ConfigType = "projectgroup"
 	ConfigTypeProject      ConfigType = "project"
 	ConfigTypeRemoteSource ConfigType = "remotesource"
@@ -51,6 +52,13 @@ func IsValidVisibility(v Visibility) bool {
 	return true
 }
 
+type MemberRole string
+
+const (
+	MemberRoleOwner  MemberRole = "owner"
+	MemberRoleMember MemberRole = "member"
+)
+
 type Parent struct {
 	Type ConfigType `json:"type,omitempty"`
 	ID   string     `json:"id,omitempty"`
@@ -91,6 +99,17 @@ type Organization struct {
 	CreatedAt     time.Time `json:"created_at,omitempty"`
 }
 
+type OrganizationMember struct {
+	Version string `json:"version,omitempty"`
+
+	ID string `json:"id,omitempty"`
+
+	OrganizationID string `json:"organization_id,omitempty"`
+	UserID         string `json:"user_id,omitempty"`
+
+	MemberRole MemberRole `json:"member_role,omitempty"`
+}
+
 type ProjectGroup struct {
 	Version string `json:"version,omitempty"`