account_conversation.rb 3.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117
  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. after_commit :push_to_streaming_api
  17. belongs_to :account
  18. belongs_to :conversation
  19. belongs_to :last_status, class_name: 'Status'
  20. before_validation :set_last_status
  21. def participant_account_ids=(arr)
  22. self[:participant_account_ids] = arr.sort
  23. end
  24. def participant_accounts
  25. if participant_account_ids.empty?
  26. [account]
  27. else
  28. participants = Account.where(id: participant_account_ids)
  29. participants.empty? ? [account] : participants
  30. end
  31. end
  32. class << self
  33. def paginate_by_id(limit, options = {})
  34. if options[:min_id]
  35. paginate_by_min_id(limit, options[:min_id]).reverse
  36. else
  37. paginate_by_max_id(limit, options[:max_id], options[:since_id])
  38. end
  39. end
  40. def paginate_by_min_id(limit, min_id = nil)
  41. query = order(arel_table[:last_status_id].asc).limit(limit)
  42. query = query.where(arel_table[:last_status_id].gt(min_id)) if min_id.present?
  43. query
  44. end
  45. def paginate_by_max_id(limit, max_id = nil, since_id = nil)
  46. query = order(arel_table[:last_status_id].desc).limit(limit)
  47. query = query.where(arel_table[:last_status_id].lt(max_id)) if max_id.present?
  48. query = query.where(arel_table[:last_status_id].gt(since_id)) if since_id.present?
  49. query
  50. end
  51. def add_status(recipient, status)
  52. conversation = find_or_initialize_by(account: recipient, conversation_id: status.conversation_id, participant_account_ids: participants_from_status(recipient, status))
  53. return conversation if conversation.status_ids.include?(status.id)
  54. conversation.status_ids << status.id
  55. conversation.unread = status.account_id != recipient.id
  56. conversation.save
  57. conversation
  58. rescue ActiveRecord::StaleObjectError
  59. retry
  60. end
  61. def remove_status(recipient, status)
  62. conversation = find_by(account: recipient, conversation_id: status.conversation_id, participant_account_ids: participants_from_status(recipient, status))
  63. return if conversation.nil?
  64. conversation.status_ids.delete(status.id)
  65. if conversation.status_ids.empty?
  66. conversation.destroy
  67. else
  68. conversation.save
  69. end
  70. conversation
  71. rescue ActiveRecord::StaleObjectError
  72. retry
  73. end
  74. private
  75. def participants_from_status(recipient, status)
  76. ((status.active_mentions.pluck(:account_id) + [status.account_id]).uniq - [recipient.id]).sort
  77. end
  78. end
  79. private
  80. def set_last_status
  81. self.status_ids = status_ids.sort
  82. self.last_status_id = status_ids.last
  83. end
  84. def push_to_streaming_api
  85. return if destroyed? || !subscribed_to_timeline?
  86. PushConversationWorker.perform_async(id)
  87. end
  88. def subscribed_to_timeline?
  89. Redis.current.exists("subscribed:#{streaming_channel}")
  90. end
  91. def streaming_channel
  92. "timeline:direct:#{account_id}"
  93. end
  94. end