correctly merge conversations when merging users, fixes #7246

previously, merging user accounts would result in conversations with
duplicate participants if both users were involved. additionally,
moved private conversations did not get their private hashes updated,
and private conversations between the merged users would lead to page
errors.

this fixes the behavior going forward (merging conversations if
needed) and adds a migration to fix broken conversations.

test plan:
1. create user A, user B and user C
2. create a private conversation between user A and user B
3. create a private conversation between user A and user C
4. create a private conversation between user B and user C
5. create a monologue for user A
6. create a monologue for user B
7. create a group conversation between all three users
8. merge user A into user B
9. confirm that user B now has three conversations:
 1. a monologue containing the messages from 2, 5, and 6
 2. a private conversation with C containing the messages from 3 and 4
 3. a group conversation with just user C
10. confirm that all messages formerly by A now have B as the author
11. confirm that unread counts and message counts are correct

Change-Id: I1afd824ff8ce74430f247d5b4c728937786fa37f
Reviewed-on: https://gerrit.instructure.com/8738
Tested-by: Hudson <hudson@instructure.com>
Reviewed-by: Zach Pendleton <zachp@instructure.com>

app/models/conversation.rb

@@ -19,11 +19,11 @@
 class Conversation < ActiveRecord::Base
   include SimpleTags
 
-  has_many :conversation_participants
+  has_many :conversation_participants, :dependent => :destroy
   has_many :subscribed_conversation_participants,
            :conditions => "subscribed",
            :class_name => 'ConversationParticipant'
-  has_many :conversation_messages, :order => "created_at DESC, id DESC"
+  has_many :conversation_messages, :order => "created_at DESC, id DESC", :dependent => :delete_all
 
   # see also User#messageable_users
   has_many :participants,
@@ -373,6 +373,48 @@ class Conversation < ActiveRecord::Base
     find_all_by_id(ids).each(&:sanitize_context_tags!)
   end
 
+  # if the participant list has changed, e.g. we merged user accounts
+  def regenerate_private_hash!(user_ids = nil)
+    return unless private?
+    self.private_hash = Conversation.private_hash_for(user_ids || self.participant_ids)
+    return unless private_hash_changed?
+    if existing = Conversation.find_by_private_hash(private_hash)
+      merge_into(existing)
+    else
+      save!
+    end
+  end
+
+  def self.batch_regenerate_private_hashes!(ids)
+    find(:all,
+     :select => "conversations.*, (SELECT #{connection.func(:group_concat, :user_id, ',')} FROM conversation_participants WHERE conversation_id = conversations.id) AS user_ids",
+     :conditions => {:id => ids}
+    ).each do |c|
+      c.regenerate_private_hash!(c.user_ids.split(',').map(&:to_i)) # group_concat order is arbitrary in sqlite, so we just let ruby do the sorting
+    end
+  end
+
+  def merge_into(other)
+    transaction do
+      new_participants = other.conversation_participants.inject({}){ |h,p| h[p.user_id] = p; h }
+      conversation_participants(true).each do |cp|
+        if new_cp = new_participants[cp.user_id]
+          new_cp.update_attribute(:workflow_state, cp.workflow_state) if cp.unread? || new_cp.archived?
+          cp.conversation_message_participants.update_all(["conversation_participant_id = ?", new_cp.id])
+          cp.destroy
+        else
+          cp.update_attribute(:conversation_id, other.id)
+        end
+      end
+      conversation_messages.update_all(["conversation_id = ?", other.id])
+      conversation_participants.reload # now empty ... need to make sure callbacks don't double-delete
+      other.conversation_participants(true).each do |cp|
+        cp.update_cached_data! :recalculate_count => true, :set_last_message_at => false, :regenerate_tags => false
+      end
+      destroy
+    end
+  end
+
   protected
 
   # contexts currently shared by > 50% of participants

app/models/conversation_participant.rb

@@ -23,7 +23,7 @@ class ConversationParticipant < ActiveRecord::Base
 
   belongs_to :conversation
   belongs_to :user
-  has_many :conversation_message_participants
+  has_many :conversation_message_participants, :dependent => :delete_all
   has_many :messages, :source => :conversation_message,
            :through => :conversation_message_participants,
            :select => "conversation_messages.*, conversation_message_participants.tags",
@@ -64,7 +64,8 @@ class ConversationParticipant < ActiveRecord::Base
 
   delegate :private?, :to => :conversation
 
-  before_update :update_unread_count
+  before_update :update_unread_count_for_update
+  before_destroy :update_unread_count_for_destroy
 
   attr_accessible :subscribed, :starred, :workflow_state
 
@@ -284,16 +285,38 @@ class ConversationParticipant < ActiveRecord::Base
     conversation.infer_new_tags_for(self, []).first
   end
 
+  def move_to_user(new_user)
+    conversation.conversation_messages.update_all(["author_id = ?", new_user.id], ["author_id = ?", user_id])
+    if existing = conversation.conversation_participants.find_by_user_id(new_user.id)
+      existing.update_attribute(:workflow_state, workflow_state) if unread? || existing.archived?
+      destroy
+    else
+      update_attribute :user_id, new_user.id
+    end
+    conversation.regenerate_private_hash! if private?
+  end
+
   protected
   def message_tags
     messages.map(&:tags).inject([], &:concat).uniq
   end
 
   private
-  def update_unread_count
+  def update_unread_count(direction=:up, user_id=self.user_id)
-    if workflow_state_changed? && [workflow_state, workflow_state_was].include?('unread')
+    User.update_all "unread_conversations_count = unread_conversations_count #{direction == :up ? '+' : '-'} 1, updated_at = '#{Time.now.to_s(:db)}'",
-      User.update_all "unread_conversations_count = unread_conversations_count #{workflow_state == 'unread' ? '+' : '-'} 1, updated_at = '#{Time.now.to_s(:db)}'",
+                    :id => user_id
-                      :id => user_id
+  end
+
+  def update_unread_count_for_update
+    if user_id_changed?
+      update_unread_count(:up) if unread?
+      update_unread_count(:down, user_id_was) if workflow_state_was == 'unread'
+    elsif workflow_state_changed? && [workflow_state, workflow_state_was].include?('unread')
+      update_unread_count(workflow_state == 'unread' ? :up : :down)
     end
   end
+
+  def update_unread_count_for_destroy
+    update_unread_count(:down) if unread?
+  end
 end

app/models/user.rb

@@ -842,10 +842,11 @@ class User < ActiveRecord::Base
         logger.error "migrating #{table} column user_id failed: #{e.to_s}"
       end
     end
+    all_conversations.find_each{ |c| c.move_to_user(new_user) }
     updates = {}
     ['account_users','asset_user_accesses',
       'assignment_reminders','attachments',
-      'calendar_events','collaborations','conversation_participants',
+      'calendar_events','collaborations',
       'context_module_progressions','discussion_entries','discussion_topics',
       'enrollments','group_memberships','page_comments','page_views',
       'rubric_assessments','short_messages',

db/migrate/20120216163427_fix_user_merge_conversations.rb

@@ -0,0 +1,51 @@
+class FixUserMergeConversations < ActiveRecord::Migration
+  def self.up
+    if supports_ddl_transactions?
+      commit_db_transaction
+      decrement_open_transactions while open_transactions > 0
+    end
+
+    # remove any duplicate CP's, possibly fixing private conversation hashes
+    # (which may merge it with another conversation)
+    ConversationParticipant.find_by_sql(<<-SQL).
+      SELECT conversation_participants.*
+      FROM conversation_participants, (
+        SELECT MIN(id) AS id, user_id, conversation_id
+        FROM conversation_participants
+        GROUP BY user_id, conversation_id
+        HAVING COUNT(*) > 1
+        ORDER BY conversation_id
+      ) cps2keep
+      WHERE conversation_participants.user_id = cps2keep.user_id
+        AND conversation_participants.conversation_id = cps2keep.conversation_id
+        AND conversation_participants.id <> cps2keep.id
+    SQL
+    each do |cp|
+      cp.destroy
+      cp.conversation.renegerate_private_hash! if cp.private?
+    end
+
+    # there may be a bunch more private conversations with the wrong private
+    # hash, and there's not a reliable way to figure out which ones those are
+    # in sql alone (unless you have a sha1 method for postgres and sqlite), so
+    # we just walk them all out of band and make sure they're right (this may
+    # also merge some private conversations in the process)
+    Conversation.connection.select_all("SELECT id FROM conversations WHERE private_hash IS NOT NULL").
+    map{ |r| r["id"] }.each_slice(1000) do |ids|
+      Conversation.send_later_if_production_enqueue_args(:batch_regenerate_private_hashes!, {
+        :priority => Delayed::LOWER_PRIORITY,
+        :max_attempts => 1,
+        :strand => "regenerate_conversation_private_hashes"
+      }, ids)
+    end
+
+    if supports_ddl_transactions?
+      increment_open_transactions
+      begin_db_transaction
+    end
+  end
+
+  def self.down
+    raise ActiveRecord::IrreversibleMigration
+  end
+end

spec/models/conversation_participant_spec.rb

@@ -152,4 +152,144 @@ describe ConversationParticipant do
       @me.conversations.tagged(@u1.asset_string, "course_1", :mode => :and).sort_by(&:id).should eql [@c8]
     end
   end
+
+  context "move_to_user" do
+    before do
+      @user1 = user_model
+      @user2 = user_model
+    end
+
+    it "should move a group conversation to the new user" do
+      c = @user1.initiate_conversation([user.id, user.id])
+      c.add_message("hello")
+      c.update_attribute(:workflow_state, 'unread')
+
+      c.move_to_user @user2
+
+      c.reload.user_id.should eql @user2.id
+      c.conversation.participant_ids.should_not include(@user1.id)
+      @user1.reload.unread_conversations_count.should eql 0
+      @user2.reload.unread_conversations_count.should eql 1
+    end
+
+    it "should clean up group conversations having both users" do
+      c = @user1.initiate_conversation([@user2.id, user.id, user.id])
+      c.add_message("hello")
+      c.update_attribute(:workflow_state, 'unread')
+      rconvo = c.conversation
+      rconvo.participant_ids.size.should eql 4
+
+      c.move_to_user @user2
+
+      lambda{ c.reload }.should raise_error # deleted
+
+      rconvo.reload
+      rconvo.participants.size.should eql 3
+      rconvo.participant_ids.should_not include(@user1.id)
+      rconvo.participant_ids.should include(@user2.id)
+      @user1.reload.unread_conversations_count.should eql 0
+      @user2.reload.unread_conversations_count.should eql 1
+    end
+
+    it "should move a private conversation to the new user" do
+      c = @user1.initiate_conversation([user.id])
+      c.add_message("hello")
+      c.update_attribute(:workflow_state, 'unread')
+      rconvo = c.conversation
+      old_hash = rconvo.private_hash
+
+      c.move_to_user @user2
+
+      c.reload.user_id.should eql @user2.id
+      rconvo.reload
+      rconvo.participants.size.should eql 2
+      rconvo.private_hash.should_not eql old_hash
+      @user1.reload.unread_conversations_count.should eql 0
+      @user2.reload.unread_conversations_count.should eql 1
+    end
+
+    it "should merge a private conversation into the existing private conversation" do
+      other_guy = user
+      c = @user1.initiate_conversation([other_guy.id])
+      c.add_message("hello")
+      c.update_attribute(:workflow_state, 'unread')
+      c2 = @user2.initiate_conversation([other_guy.id])
+      c2.add_message("hola")
+
+      c.move_to_user @user2
+
+      lambda{ c.reload }.should raise_error # deleted
+      lambda{ Conversation.find(c.conversation_id) }.should raise_error # deleted
+
+      c2.reload.messages.size.should eql 2
+      c2.messages.map(&:author_id).should eql [@user2.id, @user2.id]
+      c2.message_count.should eql 2
+      c2.user_id.should eql @user2.id
+      c2.conversation.participants.size.should eql 2
+      @user1.reload.unread_conversations_count.should eql 0
+      @user2.reload.unread_conversations_count.should eql 1
+      other_guy.reload.unread_conversations_count.should eql 1
+    end
+
+    it "should change a private conversation between the two users into a monologue" do
+      c = @user1.initiate_conversation([@user2.id])
+      c.add_message("hello self")
+      c.update_attribute(:workflow_state, 'unread')
+      @user2.mark_all_conversations_as_read!
+      rconvo = c.conversation
+      old_hash = rconvo.private_hash
+
+      c.move_to_user @user2
+
+      lambda{ c.reload }.should raise_error # deleted
+      rconvo.reload
+      rconvo.participants.size.should eql 1
+      rconvo.private_hash.should_not eql old_hash
+      @user1.reload.unread_conversations_count.should eql 0
+      @user2.reload.unread_conversations_count.should eql 1
+    end
+
+    it "should merge a private conversations between the two users into the existing monologue" do
+      c = @user1.initiate_conversation([@user2.id])
+      c.add_message("hello self")
+      c.update_attribute(:workflow_state, 'unread')
+      c2 = @user2.initiate_conversation([@user2.id])
+      c2.add_message("monologue!")
+      @user2.mark_all_conversations_as_read!
+
+      c.move_to_user @user2
+
+      lambda{ c.reload }.should raise_error # deleted
+      lambda{ Conversation.find(c.conversation_id) }.should raise_error # deleted
+
+      c2.reload.messages.size.should eql 2
+      c2.messages.map(&:author_id).should eql [@user2.id, @user2.id]
+      c2.message_count.should eql 2
+      c2.user_id.should eql @user2.id
+      c2.conversation.participants.size.should eql 1
+      @user1.reload.unread_conversations_count.should eql 0
+      @user2.reload.unread_conversations_count.should eql 1
+    end
+
+    it "should merge a monologue into the existing monologue" do
+      c = @user1.initiate_conversation([@user1.id])
+      c.add_message("monologue 1")
+      c.update_attribute(:workflow_state, 'unread')
+      c2 = @user2.initiate_conversation([@user2.id])
+      c2.add_message("monologue 2")
+
+      c.move_to_user @user2
+
+      lambda{ c.reload }.should raise_error # deleted
+      lambda{ Conversation.find(c.conversation_id) }.should raise_error # deleted
+
+      c2.reload.messages.size.should eql 2
+      c2.messages.map(&:author_id).should eql [@user2.id, @user2.id]
+      c2.message_count.should eql 2
+      c2.user_id.should eql @user2.id
+      c2.conversation.participants.size.should eql 1
+      @user1.reload.unread_conversations_count.should eql 0
+      @user2.reload.unread_conversations_count.should eql 1
+    end
+  end
 end

spec/models/user_spec.rb

@@ -533,6 +533,29 @@ describe User do
       @user1.associated_accounts.map(&:id).sort.should == []
       @user2.associated_accounts.map(&:id).sort.should == [@account1, @account2, @subaccount1, @subaccount2, @subsubaccount1, @subsubaccount2].map(&:id).sort
     end
+
+    it "should move conversations to the new user" do
+      @user1 = user_model
+      @user2 = user_model
+      c1 = @user1.initiate_conversation([user.id, user.id]) # group conversation
+      c1.add_message("hello")
+      c1.update_attribute(:workflow_state, 'unread')
+      c2 = @user1.initiate_conversation([user.id]) # private conversation
+      c2.add_message("hello")
+      c2.update_attribute(:workflow_state, 'unread')
+      old_private_hash = c2.conversation.private_hash
+
+      @user1.move_to_user @user2
+
+      c1.reload.user_id.should eql @user2.id
+      c1.conversation.participant_ids.should_not include(@user1.id)
+      @user1.reload.unread_conversations_count.should eql 0
+
+      c2.reload.user_id.should eql @user2.id
+      c2.conversation.participant_ids.should_not include(@user1.id)
+      c2.conversation.private_hash.should_not eql old_private_hash
+      @user2.reload.unread_conversations_count.should eql 2
+    end
   end
 
   context "permissions" do