3 files changed, 126 insertions, 2 deletions
diff --git a/app/models/account_conversation.rb b/app/models/account_conversation.rb
new file mode 100644
index 000000000..a7205ec1a
--- /dev/null
+++ b/app/models/account_conversation.rb
@@ -0,0 +1,111 @@
+# frozen_string_literal: true
+# == Schema Information
+#
+# Table name: account_conversations
+#
+#  id                      :bigint(8)        not null, primary key
+#  account_id              :bigint(8)
+#  conversation_id         :bigint(8)
+#  participant_account_ids :bigint(8)        default([]), not null, is an Array
+#  status_ids              :bigint(8)        default([]), not null, is an Array
+#  last_status_id          :bigint(8)
+#  lock_version            :integer          default(0), not null
+#
+
+class AccountConversation < ApplicationRecord
+  after_commit :push_to_streaming_api
+
+  belongs_to :account
+  belongs_to :conversation
+  belongs_to :last_status, class_name: 'Status'
+
+  before_validation :set_last_status
+
+  def participant_account_ids=(arr)
+    self[:participant_account_ids] = arr.sort
+  end
+
+  def participant_accounts
+    if participant_account_ids.empty?
+      [account]
+    else
+      Account.where(id: participant_account_ids)
+    end
+  end
+
+  class << self
+    def paginate_by_id(limit, options = {})
+      if options[:min_id]
+        paginate_by_min_id(limit, options[:min_id]).reverse
+      else
+        paginate_by_max_id(limit, options[:max_id], options[:since_id])
+      end
+    end
+
+    def paginate_by_min_id(limit, min_id = nil)
+      query = order(arel_table[:last_status_id].asc).limit(limit)
+      query = query.where(arel_table[:last_status_id].gt(min_id)) if min_id.present?
+      query
+    end
+
+    def paginate_by_max_id(limit, max_id = nil, since_id = nil)
+      query = order(arel_table[:last_status_id].desc).limit(limit)
+      query = query.where(arel_table[:last_status_id].lt(max_id)) if max_id.present?
+      query = query.where(arel_table[:last_status_id].gt(since_id)) if since_id.present?
+      query
+    end
+
+    def add_status(recipient, status)
+      conversation = find_or_initialize_by(account: recipient, conversation_id: status.conversation_id, participant_account_ids: participants_from_status(recipient, status))
+      conversation.status_ids << status.id
+      conversation.save
+      conversation
+    rescue ActiveRecord::StaleObjectError
+      retry
+    end
+
+    def remove_status(recipient, status)
+      conversation = find_by(account: recipient, conversation_id: status.conversation_id, participant_account_ids: participants_from_status(recipient, status))
+
+      return if conversation.nil?
+
+      conversation.status_ids.delete(status.id)
+
+      if conversation.status_ids.empty?
+        conversation.destroy
+      else
+        conversation.save
+      end
+
+      conversation
+    rescue ActiveRecord::StaleObjectError
+      retry
+    end
+
+    private
+
+    def participants_from_status(recipient, status)
+      ((status.mentions.pluck(:account_id) + [status.account_id]).uniq - [recipient.id]).sort
+    end
+  end
+
+  private
+
+  def set_last_status
+    self.status_ids     = status_ids.sort
+    self.last_status_id = status_ids.last
+  end
+
+  def push_to_streaming_api
+    return if destroyed? || !subscribed_to_timeline?
+    PushConversationWorker.perform_async(id)
+  end
+
+  def subscribed_to_timeline?
+    Redis.current.exists("subscribed:#{streaming_channel}")
+  end
+
+  def streaming_channel
+    "timeline:direct:#{account_id}"
+  end
+end
diff --git a/app/models/concerns/omniauthable.rb b/app/models/concerns/omniauthable.rb
index 50288e700..f263fe7af 100644
--- a/app/models/concerns/omniauthable.rb
+++ b/app/models/concerns/omniauthable.rb
@@ -26,7 +26,7 @@ module Omniauthable
       # to prevent the identity being locked with accidentally created accounts.
       # Note that this may leave zombie accounts (with no associated identity) which
       # can be cleaned up at a later date.
-      user = signed_in_resource ? signed_in_resource : identity.user
+      user = signed_in_resource || identity.user
       user = create_for_oauth(auth) if user.nil?
 
       if identity.user.nil?
@@ -61,7 +61,7 @@ module Omniauthable
       display_name      = auth.info.full_name || [auth.info.first_name, auth.info.last_name].join(' ')
 
       {
-        email: email ? email : "#{TEMP_EMAIL_PREFIX}-#{auth.uid}-#{auth.provider}.com",
+        email: email || "#{TEMP_EMAIL_PREFIX}-#{auth.uid}-#{auth.provider}.com",
         password: Devise.friendly_token[0, 20],
         account_attributes: {
           username: ensure_unique_username(auth.uid),
diff --git a/app/models/status.rb b/app/models/status.rb
index 028927cc3..ad25cc8df 100644
--- a/app/models/status.rb
+++ b/app/models/status.rb
@@ -26,6 +26,8 @@
 #
 
 class Status < ApplicationRecord
+  before_destroy :unlink_from_conversations
+
   include Paginable
   include Streamable
   include Cacheable
@@ -499,4 +501,15 @@ class Status < ApplicationRecord
     reblog&.decrement_count!(:reblogs_count) if reblog?
     thread&.decrement_count!(:replies_count) if in_reply_to_id.present? && (public_visibility? || unlisted_visibility?)
   end
+
+  def unlink_from_conversations
+    return unless direct_visibility?
+
+    mentioned_accounts = mentions.includes(:account).map(&:account)
+    inbox_owners       = mentioned_accounts.select(&:local?) + (account.local? ? [account] : [])
+
+    inbox_owners.each do |inbox_owner|
+      AccountConversation.remove_status(inbox_owner, self)
+    end
+  end
 end