account_conversation.rb 3.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120
  1. # frozen_string_literal: true
  2. # == Schema Information
  3. #
  4. # Table name: account_conversations
  5. #
  6. # id :bigint(8) not null, primary key
  7. # account_id :bigint(8)
  8. # conversation_id :bigint(8)
  9. # participant_account_ids :bigint(8) default([]), not null, is an Array
  10. # status_ids :bigint(8) default([]), not null, is an Array
  11. # last_status_id :bigint(8)
  12. # lock_version :integer default(0), not null
  13. # unread :boolean default(FALSE), not null
  14. #
  15. class AccountConversation < ApplicationRecord
  16. include Redisable
  17. after_commit :push_to_streaming_api
  18. belongs_to :account
  19. belongs_to :conversation
  20. belongs_to :last_status, class_name: 'Status'
  21. before_validation :set_last_status
  22. def participant_account_ids=(arr)
  23. self[:participant_account_ids] = arr.sort
  24. end
  25. def participant_accounts
  26. if participant_account_ids.empty?
  27. [account]
  28. else
  29. participants = Account.where(id: participant_account_ids)
  30. participants.empty? ? [account] : participants
  31. end
  32. end
  33. class << self
  34. def to_a_paginated_by_id(limit, options = {})
  35. if options[:min_id]
  36. paginate_by_min_id(limit, options[:min_id], options[:max_id]).reverse
  37. else
  38. paginate_by_max_id(limit, options[:max_id], options[:since_id]).to_a
  39. end
  40. end
  41. def paginate_by_min_id(limit, min_id = nil, max_id = nil)
  42. query = order(arel_table[:last_status_id].asc).limit(limit)
  43. query = query.where(arel_table[:last_status_id].gt(min_id)) if min_id.present?
  44. query = query.where(arel_table[:last_status_id].lt(max_id)) if max_id.present?
  45. query
  46. end
  47. def paginate_by_max_id(limit, max_id = nil, since_id = nil)
  48. query = order(arel_table[:last_status_id].desc).limit(limit)
  49. query = query.where(arel_table[:last_status_id].lt(max_id)) if max_id.present?
  50. query = query.where(arel_table[:last_status_id].gt(since_id)) if since_id.present?
  51. query
  52. end
  53. def add_status(recipient, status)
  54. conversation = find_or_initialize_by(account: recipient, conversation_id: status.conversation_id, participant_account_ids: participants_from_status(recipient, status))
  55. return conversation if conversation.status_ids.include?(status.id)
  56. conversation.status_ids << status.id
  57. conversation.unread = status.account_id != recipient.id
  58. conversation.save
  59. conversation
  60. rescue ActiveRecord::StaleObjectError
  61. retry
  62. end
  63. def remove_status(recipient, status)
  64. conversation = find_by(account: recipient, conversation_id: status.conversation_id, participant_account_ids: participants_from_status(recipient, status))
  65. return if conversation.nil?
  66. conversation.status_ids.delete(status.id)
  67. if conversation.status_ids.empty?
  68. conversation.destroy
  69. else
  70. conversation.save
  71. end
  72. conversation
  73. rescue ActiveRecord::StaleObjectError
  74. retry
  75. end
  76. private
  77. def participants_from_status(recipient, status)
  78. ((status.active_mentions.pluck(:account_id) + [status.account_id]).uniq - [recipient.id]).sort
  79. end
  80. end
  81. private
  82. def set_last_status
  83. self.status_ids = status_ids.sort
  84. self.last_status_id = status_ids.last
  85. end
  86. def push_to_streaming_api
  87. return if destroyed? || !subscribed_to_timeline?
  88. PushConversationWorker.perform_async(id)
  89. end
  90. def subscribed_to_timeline?
  91. redis.exists?("subscribed:#{streaming_channel}")
  92. end
  93. def streaming_channel
  94. "timeline:direct:#{account_id}"
  95. end
  96. end