canvas-lms/lib/notification_message_creator.rb

# frozen_string_literal: true

#
# Copyright (C) 2013 - present Instructure, Inc.
#
# This file is part of Canvas.
#
# Canvas is free software: you can redistribute it and/or modify it under
# the terms of the GNU Affero General Public License as published by the Free
# Software Foundation, version 3 of the License.
#
# Canvas is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
# details.
#
# You should have received a copy of the GNU Affero General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>.
#

# This file creates notification messages. I hope that was already known.
# Users have multiple communication_channels and notification preferences at
# different levels. This file accounts for these details.
#
# There are three main types of messages that are created here:
# immediate_message, delayed_messages, and dashboard_messages.
#
class NotificationMessageCreator
  include LocaleSelection

  attr_accessor :notification, :asset, :to_user_channels, :message_data
  attr_reader :courses, :account

  # Options can include:
  #  :to_list - A list of Users, User IDs, and CommunicationChannels to send to
  #  :data - Options merged with Message options
  def initialize(notification, asset, options = {})
    @notification = notification
    @asset = asset
    @to_user_channels = user_channels(options[:to_list])
    @user_counts = recent_messages_for_users(@to_user_channels.keys)
    @message_data = options.delete(:data)
    course_ids = @message_data&.dig(:course_ids)
    course_ids ||= [@message_data&.dig(:course_id)]
    root_account_id = @message_data&.dig(:root_account_id)
    if course_ids.any? && root_account_id
      @account = Account.find_cached(root_account_id)
      @courses = course_ids.map { |id| Course.new(id: id, root_account_id: @account&.id) }
    end
  end

  # Public: create (and dispatch, and queue delayed) a message
  # for this notification, associated with the given asset, sent to the given recipients
  #
  # asset - what the message applies to. An assignment, a discussion, etc.
  # to_list - a list of who to send the message to. the list can contain Users, User ids, or CommunicationChannels
  # options - a hash of extra options to merge with the options used to build the Message
  #
  # Returns a list of the messages dispatched immediately
  def create_message
    dashboard_messages = []
    immediate_messages = []
    delayed_messages = []

    @to_user_channels.each do |user, channels|
      # asset_applied_to is used for the asset (ie assignment, announcement)
      # to filter users out that do not apply to the notification like when a due
      # date is different for a specific user when using variable due dates.
      next unless (asset = asset_applied_to(user))

      user_locale = infer_locale(user: user, context: user_asset_context(asset), ignore_browser_locale: true)
      I18n.with_locale(user_locale) do
        # the channels in this method are all the users active channels or the
        # channels that were provided in the to_list.
        #
        # If the notification has an immediate_policy, it will create an
        # immediate_message or a delayed_message via build_fallback_for.
        # otherwise it will create a delayed_message. Any message can create a
        # dashboard message in addition to itself.
        channels.each do |channel|
          channel.set_root_account_ids(persist_changes: true, log: true)
          next unless notifications_enabled_for_courses?(user)

          if immediate_policy?(user, channel)
            immediate_messages << build_immediate_message_for(user, channel)
            delayed_messages << build_fallback_for(user, channel)
          else
            delayed_messages << build_delayed_message_for(user, channel)
          end
        end

        dashboard_messages << build_dashboard_message_for(user)
      end
    end
    [delayed_messages, dashboard_messages, immediate_messages].each(&:compact!)
    delayed_messages.each(&:save!)
    dispatch_immediate_messages(immediate_messages) + dispatch_dashboard_messages(dashboard_messages)
  end

  private

  # Notifications are enabled for a user in a course by default, but can be
  # disabled for notifications. The broadcast_policy needs to pass both the
  # course_id and the root_account_id to the set_broadcast_policy block for us
  # to be able to look up if it should be disabled. root_account_id is used
  # right now to look up the feature flag, but it can also be used to set
  # root_account_id on the message, or look up policy overrides in the future.
  # A user can disable notifications for a course with a notification policy
  # override.
  def notifications_enabled_for_courses?(user)
    # if the message is not summarizable?, it is in a context that notifications
    # cannot be disabled, so return true before checking.
    return true unless @notification.summarizable?

    return NotificationPolicyOverride.enabled_for_all_contexts(user, courses) if courses&.any?

    true
  end

  # fallback message is a summary message for email only that we create when we
  # have sent too many immediate messages to a user in a day.
  # returns delayed_message or nil
  def build_fallback_for(user, channel)
    # delayed_messages are only sent to email channels.
    # some types of notifications are only for immediate.
    return unless @notification.summarizable? && channel.path_type == "email"
    # we only send fallback when we did not send an immediate message, ie.
    # when the channel is bouncing or there have been too_many_messages
    return unless channel.bouncing? || too_many_messages_for?(user)

    fallback_policy = nil
    NotificationPolicy.unique_constraint_retry do
      # notification_policies are already loaded, so use find instead of generating a query
      fallback_policy = channel.notification_policies.find { |np| np.frequency == "daily" && np.notification_id.nil? }
      fallback_policy ||= channel.notification_policies.create!(frequency: "daily")
    end

    InstStatsd::Statsd.increment("message.fall_back_used", short_stat: "message.fall_back_used")
    build_summary_for(user, fallback_policy)
  end

  # returns delayed_message or nil
  def build_delayed_message_for(user, channel)
    # delayed_messages are only sent to email channels.
    # some types of notifications are only for immediate.
    return unless @notification.summarizable? && channel.path_type == "email"

    policy = effective_policy_for(user, channel)
    # if the policy is not daily or weekly, it is either immediate which was
    # picked up before in build_immediate_message_for, or it's never.
    return unless %w[daily weekly].include?(policy&.frequency)

    build_summary_for(user, policy) if policy
  end

  def build_summary_for(user, policy)
    user.shard.activate do
      message = user.messages.build(message_options_for(user))
      message.parse!("summary", root_account: @account)
      delayed_message = policy.delayed_messages.build(notification: @notification,
                                                      frequency: policy.frequency,
                                                      # policy.communication_channel should
                                                      # already be loaded in memory as the
                                                      # inverse association of loading the
                                                      # policy from the channel. passing the
                                                      # object through here lets the delayed
                                                      # message use it without having to re-query.
                                                      communication_channel: policy.communication_channel,
                                                      root_account_id: message.context_root_account.try(:id),
                                                      name_of_topic: message.subject,
                                                      link: message.url,
                                                      summary: message.body)
      delayed_message.context = @asset
      delayed_message.save! if Rails.env.test?
      delayed_message
    end
  end

  def build_immediate_message_for(user, channel)
    # if we have already created a fallback message, we don't want to make an
    # immediate message.
    return if @notification.summarizable? && too_many_messages_for?(user) && ["email", "sms"].include?(channel.path_type)

    message_options = message_options_for(user)
    message = user.messages.build(message_options.merge(communication_channel: channel, to: channel.path))
    message&.parse!(root_account: @account)
    message.workflow_state = "bounced" if channel.bouncing?
    message
  end

  def dispatch_immediate_messages(messages)
    Message.transaction do
      # Cancel any that haven't been sent out for the same purpose
      cancel_pending_duplicate_messages if Rails.env.production?
      messages.each do |message|
        message.stage_without_dispatch!
        message.save!
      end
    end
    # we filter out bounced messages now that they have been saved.
    messages = messages.select(&:staged?)
    MessageDispatcher.batch_dispatch(messages)
    messages
  end

  # returns a message or nil
  def build_dashboard_message_for(user)
    # Dashboard messages are only built if the user has finished registration,
    # if a user has never logged in, let's not spam the dashboard for no reason.
    return unless @notification.dashboard? && @notification.show_in_feed? && !user.pre_registered?

    message = user.messages.build(message_options_for(user).merge(to: "dashboard"))
    message.parse!(root_account: @account)
    message
  end

  def dispatch_dashboard_messages(messages)
    messages.each do |message|
      message.infer_defaults
      message.create_stream_items
    end
    messages
  end

  def effective_policy_for(user, channel)
    # a user can override the notification preference for a course or a
    # root_account, the course_id needs to be provided in the notification from
    # broadcast_policy in message_data, the lowest level override is the one
    # that should be respected.
    policy = override_policy_for(channel, "Course")
    policy ||= override_policy_for(channel, "Account")
    if !policy && should_use_default_policy?(user, channel)
      begin
        policy = channel.notification_policies.create!(notification_id: @notification.id, frequency: @notification.default_frequency(user))
      rescue ActiveRecord::RecordNotUnique => e
        # there is a race condition here that can happen if multiple jobs are trying to create the same
        # flavor of policy for the same user at the same time.  If we fail to save this one,
        # we should just allow the process to continue because it will find the right policy in the
        # next step of this method.
        Canvas::Errors.capture_exception(:notifications, e, :info)
      end
    end
    policy ||= channel.notification_policies.find { |np| np.notification_id == @notification.id }
    policy
  end

  def should_use_default_policy?(user, channel)
    # only use new policies for default channel when there are no other policies for the notification and user.
    # If another policy exists then it means the notification preferences page has been visited and null values
    # show as never policies in the UI.
    default_email?(user, channel) && (user.notification_policies.find { |np| np.notification_id == @notification.id }).nil?
  end

  def default_email?(user, channel)
    user.email_channel == channel
  end

  def override_policy_for(channel, context_type)
    # NotificationPolicyOverrides are already loaded and this find block is on
    # an array and can only have one for a given context and channel.
    ops = channel.notification_policy_overrides.select { |np| np.notification_id == @notification.id && np.context_type == context_type }
    case context_type
    when "Course"
      ops.find { |np| courses.map(&:id).include?(np.context_id) } if courses&.any?
    when "Account"
      ops.find { |np| np.context_id == account.id } if account
    end
  end

  def user_channels(to_list)
    to_user_channels = Hash.new([])
    # if this method is given users we preload communication channels and they
    # are already loaded so we are using the select :active? to not do another
    # query to load them again.
    users_from_to_list(to_list).each do |user|
      to_user_channels[user] += user.communication_channels.select { |cc| add_channel?(user, cc) }
    end
    # if the method gets communication channels, the user is loaded, and this
    # allows all the methods in this file to behave the same as if it were users.
    communication_channels_from_to_list(to_list).each do |channel|
      to_user_channels[channel.user] += [channel]
    end
    to_user_channels.each_value(&:uniq!)
    to_user_channels
  end

  # only send emails to active channels or registration notifications to default users' channel
  def add_channel?(user, channel)
    return false if channel.path_type == CommunicationChannel::TYPE_SMS

    channel.active? || (@notification.registration? && default_email?(user, channel))
  end

  def users_from_to_list(to_list)
    to_list = [to_list] unless to_list.is_a? Enumerable

    to_users = []
    to_users += User.find(to_list.select { |to| to.is_a? Numeric }.uniq)
    to_users += to_list.select { |to| to.is_a? User }
    to_users.uniq!

    to_users
  end

  def communication_channels_from_to_list(to_list)
    to_list = [to_list] unless to_list.is_a? Enumerable
    to_list.select { |to| to.is_a? CommunicationChannel }.uniq
  end

  def asset_applied_to(user)
    if asset.respond_to?(:filter_asset_by_recipient)
      asset.filter_asset_by_recipient(@notification, user)
    else
      asset
    end
  end

  def message_options_for(user)
    user_asset = asset_applied_to(user)

    message_options = {
      subject: @notification.subject,
      notification: @notification,
      notification_name: @notification.name,
      user: user,
      context: user_asset,
    }

    # can't just merge these because nil values need to be overwritten in a later merge
    message_options[:delay_for] = @notification.delay_for if @notification.delay_for
    message_options[:data] = @message_data if @message_data
    message_options
  end

  def user_asset_context(user_asset)
    if user_asset.is_a?(Context)
      user_asset
    elsif user_asset.respond_to?(:context)
      user_asset.context
    end
  end

  def immediate_policy?(user, channel)
    # we want to ignore unconfirmed channels unless the notification is
    # registration because that is how the user can confirm the channel
    return true if @notification.registration?
    # pre_registered users should only get registration emails.
    return false if user.pre_registered?
    return false if channel.unconfirmed?
    return true if @notification.migration?

    policy = effective_policy_for(user, channel)
    policy&.frequency == "immediately"
  end

  def cancel_pending_duplicate_messages
    first_start_time = start_time = Setting.get("pending_duplicate_message_window_hours", "6").to_i.hours.ago
    final_end_time = Time.now.utc
    first_partition = Message.infer_partition_table_name("created_at" => first_start_time)

    loop do
      end_time = start_time + 7.days
      end_time = final_end_time if end_time > final_end_time
      scope = Message
              .in_partition("created_at" => start_time)
              .where(notification_id: @notification)
              .for(@asset)
              .by_name(@notification.name)
              .for_user(@to_user_channels.keys)
              .cancellable
      start_partition = Message.infer_partition_table_name("created_at" => start_time)
      end_partition = Message.infer_partition_table_name("created_at" => end_time)
      if first_partition == start_partition &&
         start_partition == end_partition
        scope = scope.where(created_at: start_time..end_time)
        break_this_loop = true
      elsif start_time == first_start_time
        scope = scope.where("created_at>=?", start_time)
      elsif start_partition == end_partition
        scope = scope.where("created_at<=?", end_time)
        break_this_loop = true
        # else <no conditions; we're addressing the entire partition>
      end
      scope.update_all(workflow_state: "cancelled") if Message.connection.table_exists?(start_partition)

      break if break_this_loop

      start_time = end_time
    end
  end

  def too_many_messages_for?(user)
    if @user_counts[user.id] >= user.max_messages_per_day
      InstStatsd::Statsd.increment("message.too_many_messages_for_was_true", short_stat: "message.too_many_messages_for_was_true")
      true
    end
  end

  # Cache the count for number of messages sent to a user/user-with-category,
  # it can also be manually re-set to reflect new rows added... this cache
  # data can get out of sync if messages are cancelled for being repeats...
  # not sure if we care about that...
  def recent_messages_for_users(users)
    GuardRail.activate(:secondary) do
      Hash.new(0).merge(Message.more_recent_than(24.hours.ago).where(user_id: users, to_email: true).group(:user_id).count)
    end
  end
end