account_conversation.rb 3.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118
  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 to_a_paginated_by_id(limit, options = {})
  34. if options[:min_id]
  35. paginate_by_min_id(limit, options[:min_id], options[:max_id]).reverse
  36. else
  37. paginate_by_max_id(limit, options[:max_id], options[:since_id]).to_a
  38. end
  39. end
  40. def paginate_by_min_id(limit, min_id = nil, max_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 = query.where(arel_table[:last_status_id].lt(max_id)) if max_id.present?
  44. query
  45. end
  46. def paginate_by_max_id(limit, max_id = nil, since_id = nil)
  47. query = order(arel_table[:last_status_id].desc).limit(limit)
  48. query = query.where(arel_table[:last_status_id].lt(max_id)) if max_id.present?
  49. query = query.where(arel_table[:last_status_id].gt(since_id)) if since_id.present?
  50. query
  51. end
  52. def add_status(recipient, status)
  53. conversation = find_or_initialize_by(account: recipient, conversation_id: status.conversation_id, participant_account_ids: participants_from_status(recipient, status))
  54. return conversation if conversation.status_ids.include?(status.id)
  55. conversation.status_ids << status.id
  56. conversation.unread = status.account_id != recipient.id
  57. conversation.save
  58. conversation
  59. rescue ActiveRecord::StaleObjectError
  60. retry
  61. end
  62. def remove_status(recipient, status)
  63. conversation = find_by(account: recipient, conversation_id: status.conversation_id, participant_account_ids: participants_from_status(recipient, status))
  64. return if conversation.nil?
  65. conversation.status_ids.delete(status.id)
  66. if conversation.status_ids.empty?
  67. conversation.destroy
  68. else
  69. conversation.save
  70. end
  71. conversation
  72. rescue ActiveRecord::StaleObjectError
  73. retry
  74. end
  75. private
  76. def participants_from_status(recipient, status)
  77. ((status.active_mentions.pluck(:account_id) + [status.account_id]).uniq - [recipient.id]).sort
  78. end
  79. end
  80. private
  81. def set_last_status
  82. self.status_ids = status_ids.sort
  83. self.last_status_id = status_ids.last
  84. end
  85. def push_to_streaming_api
  86. return if destroyed? || !subscribed_to_timeline?
  87. PushConversationWorker.perform_async(id)
  88. end
  89. def subscribed_to_timeline?
  90. Redis.current.exists?("subscribed:#{streaming_channel}")
  91. end
  92. def streaming_channel
  93. "timeline:direct:#{account_id}"
  94. end
  95. end