Refactoring direct channel creation to use transaction to avaoid state where channel is created without both users

api/channel.go

@@ -131,16 +131,21 @@ func CreateDirectChannel(c *Context, otherUserId string) (*model.Channel, *model
 		return nil, model.NewAppError("CreateDirectChannel", "Invalid other user id ", otherUserId)
 	}
 
-	if sc, err := CreateChannel(c, channel, true); err != nil {
+	cm1 := &model.ChannelMember{
-		return nil, err
+		UserId:      c.Session.UserId,
+		Roles:       model.CHANNEL_ROLE_ADMIN,
+		NotifyProps: model.GetDefaultChannelNotifyProps(),
+	}
+	cm2 := &model.ChannelMember{
+		UserId:      otherUserId,
+		Roles:       "",
+		NotifyProps: model.GetDefaultChannelNotifyProps(),
+	}
+
+	if result := <-Srv.Store.Channel().SaveDirectChannel(channel, cm1, cm2); result.Err != nil {
+		return nil, result.Err
 	} else {
-		cm := &model.ChannelMember{ChannelId: sc.Id, UserId: otherUserId, Roles: "", NotifyProps: model.GetDefaultChannelNotifyProps()}
+		return result.Data.(*model.Channel), nil
-
-		if cmresult := <-Srv.Store.Channel().SaveMember(cm); cmresult.Err != nil {
-			return nil, cmresult.Err
-		}
-
-		return sc, nil
 	}
 }
 

store/sql_channel_store.go

@@ -5,6 +5,7 @@ package store
 
 import (
 	l4g "code.google.com/p/log4go"
+	"github.com/go-gorp/gorp"
 	"github.com/mattermost/platform/model"
 	"github.com/mattermost/platform/utils"
 )
@@ -97,49 +98,22 @@ func (s SqlChannelStore) Save(channel *model.Channel) StoreChannel {
 	storeChannel := make(StoreChannel)
 
 	go func() {
-		result := StoreResult{}
+		var result StoreResult
-
+		if channel.Type == model.CHANNEL_DIRECT {
-		if len(channel.Id) > 0 {
+			result.Err = model.NewAppError("SqlChannelStore.Save", "Use SaveDirectChannel to create a direct channel", "")
-			result.Err = model.NewAppError("SqlChannelStore.Save",
-				"Must call update for exisiting channel", "id="+channel.Id)
-			storeChannel <- result
-			close(storeChannel)
-			return
-		}
-
-		channel.PreSave()
-		if result.Err = channel.IsValid(); result.Err != nil {
-			storeChannel <- result
-			close(storeChannel)
-			return
-		}
-
-		if count, err := s.GetMaster().SelectInt("SELECT COUNT(0) FROM Channels WHERE TeamId = :TeamId AND DeleteAt = 0 AND (Type = 'O' OR Type = 'P')", map[string]interface{}{"TeamId": channel.TeamId}); err != nil {
-			result.Err = model.NewAppError("SqlChannelStore.Save", "Failed to get current channel count", "teamId="+channel.TeamId+", "+err.Error())
-			storeChannel <- result
-			close(storeChannel)
-			return
-		} else if count > 150 {
-			result.Err = model.NewAppError("SqlChannelStore.Save", "You've reached the limit of the number of allowed channels.", "teamId="+channel.TeamId)
-			storeChannel <- result
-			close(storeChannel)
-			return
-		}
-
-		if err := s.GetMaster().Insert(channel); err != nil {
-			if IsUniqueConstraintError(err.Error(), "Name", "channels_name_teamid_key") {
-				dupChannel := model.Channel{}
-				s.GetReplica().SelectOne(&dupChannel, "SELECT * FROM Channels WHERE TeamId = :TeamId AND Name = :Name AND DeleteAt > 0", map[string]interface{}{"TeamId": channel.TeamId, "Name": channel.Name})
-				if dupChannel.DeleteAt > 0 {
-					result.Err = model.NewAppError("SqlChannelStore.Update", "A channel with that URL was previously created", "id="+channel.Id+", "+err.Error())
-				} else {
-					result.Err = model.NewAppError("SqlChannelStore.Update", "A channel with that URL already exists", "id="+channel.Id+", "+err.Error())
-				}
-			} else {
-				result.Err = model.NewAppError("SqlChannelStore.Save", "We couldn't save the channel", "id="+channel.Id+", "+err.Error())
-			}
 		} else {
-			result.Data = channel
+			if transaction, err := s.GetMaster().Begin(); err != nil {
+				result.Err = model.NewAppError("SqlChannelStore.Save", "Unable to open transaction", err.Error())
+			} else {
+				result = s.saveChannelT(transaction, channel)
+				if result.Err != nil {
+					transaction.Rollback()
+				} else {
+					if err := transaction.Commit(); err != nil {
+						result.Err = model.NewAppError("SqlChannelStore.Save", "Unable to commit transaction", err.Error())
+					}
+				}
+			}
 		}
 
 		storeChannel <- result
@@ -149,6 +123,100 @@ func (s SqlChannelStore) Save(channel *model.Channel) StoreChannel {
 	return storeChannel
 }
 
+func (s SqlChannelStore) SaveDirectChannel(directchannel *model.Channel, member1 *model.ChannelMember, member2 *model.ChannelMember) StoreChannel {
+	storeChannel := make(StoreChannel)
+
+	go func() {
+		var result StoreResult
+
+		if directchannel.Type != model.CHANNEL_DIRECT {
+			result.Err = model.NewAppError("SqlChannelStore.SaveDirectChannel", "Not a direct channel attempted to be created with SaveDirectChannel", "")
+		} else {
+			if transaction, err := s.GetMaster().Begin(); err != nil {
+				result.Err = model.NewAppError("SqlChannelStore.SaveDirectChannel", "Unable to open transaction", err.Error())
+			} else {
+				channelResult := s.saveChannelT(transaction, directchannel)
+
+				if channelResult.Err != nil {
+					transaction.Rollback()
+					result.Err = channelResult.Err
+				} else {
+					newChannel := channelResult.Data.(*model.Channel)
+					// Members need new channel ID
+					member1.ChannelId = newChannel.Id
+					member2.ChannelId = newChannel.Id
+
+					member1Result := s.saveMemberT(transaction, member1, newChannel)
+					member2Result := s.saveMemberT(transaction, member2, newChannel)
+
+					if member1Result.Err != nil || member2Result.Err != nil {
+						transaction.Rollback()
+						details := ""
+						if member1Result.Err != nil {
+							details += "Member1Err: " + member1Result.Err.Message
+						}
+						if member2Result.Err != nil {
+							details += "Member2Err: " + member2Result.Err.Message
+						}
+						result.Err = model.NewAppError("SqlChannelStore.SaveDirectChannel", "Unable to add direct channel members", details)
+					} else {
+						if err := transaction.Commit(); err != nil {
+							result.Err = model.NewAppError("SqlChannelStore.SaveDirectChannel", "Ubable to commit transaction", err.Error())
+						} else {
+							result = channelResult
+						}
+					}
+				}
+			}
+		}
+
+		storeChannel <- result
+		close(storeChannel)
+	}()
+
+	return storeChannel
+}
+
+func (s SqlChannelStore) saveChannelT(transaction *gorp.Transaction, channel *model.Channel) StoreResult {
+	result := StoreResult{}
+
+	if len(channel.Id) > 0 {
+		result.Err = model.NewAppError("SqlChannelStore.Save", "Must call update for exisiting channel", "id="+channel.Id)
+		return result
+	}
+
+	channel.PreSave()
+	if result.Err = channel.IsValid(); result.Err != nil {
+		return result
+	}
+
+	if count, err := transaction.SelectInt("SELECT COUNT(0) FROM Channels WHERE TeamId = :TeamId AND DeleteAt = 0 AND (Type = 'O' OR Type = 'P')", map[string]interface{}{"TeamId": channel.TeamId}); err != nil {
+		result.Err = model.NewAppError("SqlChannelStore.Save", "Failed to get current channel count", "teamId="+channel.TeamId+", "+err.Error())
+		return result
+	} else if count > 150 {
+		result.Err = model.NewAppError("SqlChannelStore.Save", "You've reached the limit of the number of allowed channels.", "teamId="+channel.TeamId)
+		return result
+	}
+
+	if err := transaction.Insert(channel); err != nil {
+		if IsUniqueConstraintError(err.Error(), "Name", "channels_name_teamid_key") {
+			dupChannel := model.Channel{}
+			s.GetReplica().SelectOne(&dupChannel, "SELECT * FROM Channels WHERE TeamId = :TeamId AND Name = :Name AND DeleteAt > 0", map[string]interface{}{"TeamId": channel.TeamId, "Name": channel.Name})
+			if dupChannel.DeleteAt > 0 {
+				result.Err = model.NewAppError("SqlChannelStore.Update", "A channel with that URL was previously created", "id="+channel.Id+", "+err.Error())
+			} else {
+				result.Err = model.NewAppError("SqlChannelStore.Update", "A channel with that URL already exists", "id="+channel.Id+", "+err.Error())
+			}
+		} else {
+			result.Err = model.NewAppError("SqlChannelStore.Save", "We couldn't save the channel", "id="+channel.Id+", "+err.Error())
+		}
+	} else {
+		result.Data = channel
+	}
+
+	return result
+}
+
 func (s SqlChannelStore) Update(channel *model.Channel) StoreChannel {
 
 	storeChannel := make(StoreChannel)
@@ -396,31 +464,27 @@ func (s SqlChannelStore) SaveMember(member *model.ChannelMember) StoreChannel {
 	storeChannel := make(StoreChannel)
 
 	go func() {
-		result := StoreResult{}
+		var result StoreResult
-
 		// Grab the channel we are saving this member to
 		if cr := <-s.Get(member.ChannelId); cr.Err != nil {
 			result.Err = cr.Err
 		} else {
 			channel := cr.Data.(*model.Channel)
 
-			member.PreSave()
+			if transaction, err := s.GetMaster().Begin(); err != nil {
-			if result.Err = member.IsValid(); result.Err != nil {
+				result.Err = model.NewAppError("SqlChannelStore.SaveMember", "Unable to open transaction", err.Error())
-				storeChannel <- result
-				return
-			}
-
-			if err := s.GetMaster().Insert(member); err != nil {
-				if IsUniqueConstraintError(err.Error(), "ChannelId", "channelmembers_pkey") {
-					result.Err = model.NewAppError("SqlChannelStore.SaveMember", "A channel member with that id already exists", "channel_id="+member.ChannelId+", user_id="+member.UserId+", "+err.Error())
-				} else {
-					result.Err = model.NewAppError("SqlChannelStore.SaveMember", "We couldn't save the channel member", "channel_id="+member.ChannelId+", user_id="+member.UserId+", "+err.Error())
-				}
 			} else {
-				result.Data = member
+				result = s.saveMemberT(transaction, member, channel)
-				// If sucessfull record members have changed in channel
+				if result.Err != nil {
-				if mu := <-s.extraUpdated(channel); mu.Err != nil {
+					transaction.Rollback()
-					result.Err = mu.Err
+				} else {
+					if err := transaction.Commit(); err != nil {
+						result.Err = model.NewAppError("SqlChannelStore.SaveMember", "Unable to commit transaction", err.Error())
+					}
+					// If sucessfull record members have changed in channel
+					if mu := <-s.extraUpdated(channel); mu.Err != nil {
+						result.Err = mu.Err
+					}
 				}
 			}
 		}
@@ -432,6 +496,27 @@ func (s SqlChannelStore) SaveMember(member *model.ChannelMember) StoreChannel {
 	return storeChannel
 }
 
+func (s SqlChannelStore) saveMemberT(transaction *gorp.Transaction, member *model.ChannelMember, channel *model.Channel) StoreResult {
+	result := StoreResult{}
+
+	member.PreSave()
+	if result.Err = member.IsValid(); result.Err != nil {
+		return result
+	}
+
+	if err := transaction.Insert(member); err != nil {
+		if IsUniqueConstraintError(err.Error(), "ChannelId", "channelmembers_pkey") {
+			result.Err = model.NewAppError("SqlChannelStore.SaveMember", "A channel member with that id already exists", "channel_id="+member.ChannelId+", user_id="+member.UserId+", "+err.Error())
+		} else {
+			result.Err = model.NewAppError("SqlChannelStore.SaveMember", "We couldn't save the channel member", "channel_id="+member.ChannelId+", user_id="+member.UserId+", "+err.Error())
+		}
+	} else {
+		result.Data = member
+	}
+
+	return result
+}
+
 func (s SqlChannelStore) UpdateMember(member *model.ChannelMember) StoreChannel {
 	storeChannel := make(StoreChannel)
 

store/sql_channel_store_test.go

@@ -33,6 +33,14 @@ func TestChannelStoreSave(t *testing.T) {
 		t.Fatal("should be unique name")
 	}
 
+	o1.Id = ""
+	o1.Name = "a" + model.NewId() + "b"
+	o1.Type = model.CHANNEL_DIRECT
+	if err := (<-store.Channel().Save(&o1)).Err; err == nil {
+		t.Fatal("Should not be able to save direct channel")
+	}
+
+	o1.Type = model.CHANNEL_OPEN
 	for i := 0; i < 150; i++ {
 		o1.Id = ""
 		o1.Name = "a" + model.NewId() + "b"
@@ -48,6 +56,61 @@ func TestChannelStoreSave(t *testing.T) {
 	}
 }
 
+func TestChannelStoreSaveDirectChannel(t *testing.T) {
+	Setup()
+
+	teamId := model.NewId()
+
+	o1 := model.Channel{}
+	o1.TeamId = teamId
+	o1.DisplayName = "Name"
+	o1.Name = "a" + model.NewId() + "b"
+	o1.Type = model.CHANNEL_DIRECT
+
+	u1 := model.User{}
+	u1.TeamId = model.NewId()
+	u1.Email = model.NewId()
+	u1.Nickname = model.NewId()
+	Must(store.User().Save(&u1))
+
+	u2 := model.User{}
+	u2.TeamId = model.NewId()
+	u2.Email = model.NewId()
+	u2.Nickname = model.NewId()
+	Must(store.User().Save(&u2))
+
+	m1 := model.ChannelMember{}
+	m1.ChannelId = o1.Id
+	m1.UserId = u1.Id
+	m1.NotifyProps = model.GetDefaultChannelNotifyProps()
+
+	m2 := model.ChannelMember{}
+	m2.ChannelId = o1.Id
+	m2.UserId = u2.Id
+	m2.NotifyProps = model.GetDefaultChannelNotifyProps()
+
+	if err := (<-store.Channel().SaveDirectChannel(&o1, &m1, &m2)).Err; err != nil {
+		t.Fatal("couldn't save direct channel", err)
+	}
+
+	members := (<-store.Channel().GetMembers(o1.Id)).Data.([]model.ChannelMember)
+	if len(members) != 2 {
+		t.Fatal("should have saved 2 members")
+	}
+
+	if err := (<-store.Channel().SaveDirectChannel(&o1, &m1, &m2)).Err; err == nil {
+		t.Fatal("shouldn't be able to update from save")
+	}
+
+	o1.Id = ""
+	o1.Name = "a" + model.NewId() + "b"
+	o1.Type = model.CHANNEL_OPEN
+	if err := (<-store.Channel().SaveDirectChannel(&o1, &m1, &m2)).Err; err == nil {
+		t.Fatal("Should not be able to save non-direct channel")
+	}
+
+}
+
 func TestChannelStoreUpdate(t *testing.T) {
 	Setup()
 
@@ -99,6 +162,44 @@ func TestChannelStoreGet(t *testing.T) {
 	if err := (<-store.Channel().Get("")).Err; err == nil {
 		t.Fatal("Missing id should have failed")
 	}
+
+	u1 := model.User{}
+	u1.TeamId = model.NewId()
+	u1.Email = model.NewId()
+	u1.Nickname = model.NewId()
+	Must(store.User().Save(&u1))
+
+	u2 := model.User{}
+	u2.TeamId = model.NewId()
+	u2.Email = model.NewId()
+	u2.Nickname = model.NewId()
+	Must(store.User().Save(&u2))
+
+	o2 := model.Channel{}
+	o2.TeamId = model.NewId()
+	o2.DisplayName = "Direct Name"
+	o2.Name = "a" + model.NewId() + "b"
+	o2.Type = model.CHANNEL_DIRECT
+
+	m1 := model.ChannelMember{}
+	m1.ChannelId = o2.Id
+	m1.UserId = u1.Id
+	m1.NotifyProps = model.GetDefaultChannelNotifyProps()
+
+	m2 := model.ChannelMember{}
+	m2.ChannelId = o2.Id
+	m2.UserId = u2.Id
+	m2.NotifyProps = model.GetDefaultChannelNotifyProps()
+
+	Must(store.Channel().SaveDirectChannel(&o2, &m1, &m2))
+
+	if r2 := <-store.Channel().Get(o2.Id); r2.Err != nil {
+		t.Fatal(r2.Err)
+	} else {
+		if r2.Data.(*model.Channel).ToJson() != o2.ToJson() {
+			t.Fatal("invalid returned channel")
+		}
+	}
 }
 
 func TestChannelStoreDelete(t *testing.T) {

store/store.go

@@ -54,6 +54,7 @@ type TeamStore interface {
 
 type ChannelStore interface {
 	Save(channel *model.Channel) StoreChannel
+	SaveDirectChannel(channel *model.Channel, member1 *model.ChannelMember, member2 *model.ChannelMember) StoreChannel
 	Update(channel *model.Channel) StoreChannel
 	Get(id string) StoreChannel
 	Delete(channelId string, time int64) StoreChannel