Add experimental server-side notification grouping (#29889)

app/controllers/api/v2_alpha/notifications_controller.rb

@@ -0,0 +1,91 @@
+# frozen_string_literal: true
+
+class Api::V2Alpha::NotificationsController < Api::BaseController
+  before_action -> { doorkeeper_authorize! :read, :'read:notifications' }, except: [:clear, :dismiss]
+  before_action -> { doorkeeper_authorize! :write, :'write:notifications' }, only: [:clear, :dismiss]
+  before_action :require_user!
+  after_action :insert_pagination_headers, only: :index
+
+  DEFAULT_NOTIFICATIONS_LIMIT = 40
+
+  def index
+    with_read_replica do
+      @notifications = load_notifications
+      @group_metadata = load_group_metadata
+      @relationships = StatusRelationshipsPresenter.new(target_statuses_from_notifications, current_user&.account_id)
+    end
+
+    render json: @notifications.map { |notification| NotificationGroup.from_notification(notification) }, each_serializer: REST::NotificationGroupSerializer, relationships: @relationships, group_metadata: @group_metadata
+  end
+
+  def show
+    @notification = current_account.notifications.without_suspended.find_by!(group_key: params[:id])
+    render json: NotificationGroup.from_notification(@notification), serializer: REST::NotificationGroupSerializer
+  end
+
+  def clear
+    current_account.notifications.delete_all
+    render_empty
+  end
+
+  def dismiss
+    current_account.notifications.where(group_key: params[:id]).destroy_all
+    render_empty
+  end
+
+  private
+
+  def load_notifications
+    notifications = browserable_account_notifications.includes(from_account: [:account_stat, :user]).to_a_grouped_paginated_by_id(
+      limit_param(DEFAULT_NOTIFICATIONS_LIMIT),
+      params_slice(:max_id, :since_id, :min_id)
+    )
+
+    Notification.preload_cache_collection_target_statuses(notifications) do |target_statuses|
+      preload_collection(target_statuses, Status)
+    end
+  end
+
+  def load_group_metadata
+    return {} if @notifications.empty?
+
+    browserable_account_notifications
+      .where(group_key: @notifications.filter_map(&:group_key))
+      .where(id: (@notifications.last.id)..(@notifications.first.id))
+      .group(:group_key)
+      .pluck(:group_key, 'min(notifications.id) as min_id', 'max(notifications.id) as max_id', 'max(notifications.created_at) as latest_notification_at')
+      .to_h { |group_key, min_id, max_id, latest_notification_at| [group_key, { min_id: min_id, max_id: max_id, latest_notification_at: latest_notification_at }] }
+  end
+
+  def browserable_account_notifications
+    current_account.notifications.without_suspended.browserable(
+      types: Array(browserable_params[:types]),
+      exclude_types: Array(browserable_params[:exclude_types]),
+      include_filtered: truthy_param?(:include_filtered)
+    )
+  end
+
+  def target_statuses_from_notifications
+    @notifications.filter_map(&:target_status)
+  end
+
+  def next_path
+    api_v2_alpha_notifications_url pagination_params(max_id: pagination_max_id) unless @notifications.empty?
+  end
+
+  def prev_path
+    api_v2_alpha_notifications_url pagination_params(min_id: pagination_since_id) unless @notifications.empty?
+  end
+
+  def pagination_collection
+    @notifications
+  end
+
+  def browserable_params
+    params.permit(:include_filtered, types: [], exclude_types: [])
+  end
+
+  def pagination_params(core_params)
+    params.slice(:limit, :types, :exclude_types, :include_filtered).permit(:limit, :include_filtered, types: [], exclude_types: []).merge(core_params)
+  end
+end

app/models/notification.rb

@@ -13,6 +13,7 @@
 #  from_account_id :bigint(8)        not null
 #  type            :string
 #  filtered        :boolean          default(FALSE), not null
+#  group_key       :string
 #
 
 class Notification < ApplicationRecord
@@ -136,6 +137,67 @@ def browserable(types: [], exclude_types: [], from_account_id: nil, include_filt
       end
     end
 
+    # This returns notifications from the request page, but with at most one notification per group.
+    # Notifications that have no `group_key` each count as a separate group.
+    def paginate_groups_by_max_id(limit, max_id: nil, since_id: nil)
+      query = reorder(id: :desc)
+      query = query.where(id: ...max_id) if max_id.present?
+      query = query.where(id: (since_id + 1)...) if since_id.present?
+
+      unscoped
+        .with_recursive(
+          grouped_notifications: [
+            query
+              .select('notifications.*', "ARRAY[COALESCE(notifications.group_key, 'ungrouped-' || notifications.id)] groups")
+              .limit(1),
+            query
+              .joins('CROSS JOIN grouped_notifications')
+              .where('notifications.id < grouped_notifications.id')
+              .where.not("COALESCE(notifications.group_key, 'ungrouped-' || notifications.id) = ANY(grouped_notifications.groups)")
+              .select('notifications.*', "array_append(grouped_notifications.groups, COALESCE(notifications.group_key, 'ungrouped-' || notifications.id))")
+              .limit(1),
+          ]
+        )
+        .from('grouped_notifications AS notifications')
+        .order(id: :desc)
+        .limit(limit)
+    end
+
+    # Differs from :paginate_groups_by_max_id in that it gives the results immediately following min_id,
+    # whereas since_id gives the items with largest id, but with since_id as a cutoff.
+    # Results will be in ascending order by id.
+    def paginate_groups_by_min_id(limit, max_id: nil, min_id: nil)
+      query = reorder(id: :asc)
+      query = query.where(id: (min_id + 1)...) if min_id.present?
+      query = query.where(id: ...max_id) if max_id.present?
+
+      unscoped
+        .with_recursive(
+          grouped_notifications: [
+            query
+              .select('notifications.*', "ARRAY[COALESCE(notifications.group_key, 'ungrouped-' || notifications.id)] groups")
+              .limit(1),
+            query
+              .joins('CROSS JOIN grouped_notifications')
+              .where('notifications.id > grouped_notifications.id')
+              .where.not("COALESCE(notifications.group_key, 'ungrouped-' || notifications.id) = ANY(grouped_notifications.groups)")
+              .select('notifications.*', "array_append(grouped_notifications.groups, COALESCE(notifications.group_key, 'ungrouped-' || notifications.id))")
+              .limit(1),
+          ]
+        )
+        .from('grouped_notifications AS notifications')
+        .order(id: :asc)
+        .limit(limit)
+    end
+
+    def to_a_grouped_paginated_by_id(limit, options = {})
+      if options[:min_id].present?
+        paginate_groups_by_min_id(limit, min_id: options[:min_id], max_id: options[:max_id]).reverse
+      else
+        paginate_groups_by_max_id(limit, max_id: options[:max_id], since_id: options[:since_id]).to_a
+      end
+    end
+
     def preload_cache_collection_target_statuses(notifications, &_block)
       notifications.group_by(&:type).each do |type, grouped_notifications|
         associations = TARGET_STATUS_INCLUDES_BY_TYPE[type]

app/models/notification_group.rb

@@ -0,0 +1,29 @@
+# frozen_string_literal: true
+
+class NotificationGroup < ActiveModelSerializers::Model
+  attributes :group_key, :sample_accounts, :notifications_count, :notification
+
+  def self.from_notification(notification)
+    if notification.group_key.present?
+      # TODO: caching and preloading
+      sample_accounts = notification.account.notifications.where(group_key: notification.group_key).order(id: :desc).limit(3).map(&:from_account)
+      notifications_count = notification.account.notifications.where(group_key: notification.group_key).count
+    else
+      sample_accounts = [notification.from_account]
+      notifications_count = 1
+    end
+
+    NotificationGroup.new(
+      notification: notification,
+      group_key: notification.group_key || "ungrouped-#{notification.id}",
+      sample_accounts: sample_accounts,
+      notifications_count: notifications_count
+    )
+  end
+
+  delegate :type,
+           :target_status,
+           :report,
+           :account_relationship_severance_event,
+           to: :notification, prefix: false
+end

app/serializers/rest/notification_group_serializer.rb

@@ -0,0 +1,45 @@
+# frozen_string_literal: true
+
+class REST::NotificationGroupSerializer < ActiveModel::Serializer
+  attributes :group_key, :notifications_count, :type
+
+  attribute :page_min_id, if: :paginated?
+  attribute :page_max_id, if: :paginated?
+  attribute :latest_page_notification_at, if: :paginated?
+
+  has_many :sample_accounts, serializer: REST::AccountSerializer
+  belongs_to :target_status, key: :status, if: :status_type?, serializer: REST::StatusSerializer
+  belongs_to :report, if: :report_type?, serializer: REST::ReportSerializer
+  belongs_to :account_relationship_severance_event, key: :event, if: :relationship_severance_event?, serializer: REST::AccountRelationshipSeveranceEventSerializer
+
+  def status_type?
+    [:favourite, :reblog, :status, :mention, :poll, :update].include?(object.type)
+  end
+
+  def report_type?
+    object.type == :'admin.report'
+  end
+
+  def relationship_severance_event?
+    object.type == :severed_relationships
+  end
+
+  def page_min_id
+    range = instance_options[:group_metadata][object.group_key]
+    range.present? ? range[:min_id].to_s : object.notification.id.to_s
+  end
+
+  def page_max_id
+    range = instance_options[:group_metadata][object.group_key]
+    range.present? ? range[:max_id].to_s : object.notification.id.to_s
+  end
+
+  def latest_page_notification_at
+    range = instance_options[:group_metadata][object.group_key]
+    range.present? ? range[:latest_notification_at] : object.notification.created_at
+  end
+
+  def paginated?
+    instance_options[:group_metadata].present?
+  end
+end

app/services/notify_service.rb

@@ -3,6 +3,9 @@
 class NotifyService < BaseService
   include Redisable
 
+  MAXIMUM_GROUP_SPAN_HOURS = 12
+  MAXIMUM_GROUP_GAP_TIME = 4.hours.to_i
+
   NON_EMAIL_TYPES = %i(
     admin.report
     admin.sign_up
@@ -183,6 +186,7 @@ def call(recipient, type, activity)
     return if dismiss?
 
     @notification.filtered = filter?
+    @notification.group_key = notification_group_key
     @notification.save!
 
     # It's possible the underlying activity has been deleted
@@ -202,6 +206,24 @@ def call(recipient, type, activity)
 
   private
 
+  def notification_group_key
+    return nil if @notification.filtered || %i(favourite reblog).exclude?(@notification.type)
+
+    type_prefix = "#{@notification.type}-#{@notification.target_status.id}"
+    redis_key   = "notif-group/#{@recipient.id}/#{type_prefix}"
+    hour_bucket = @notification.activity.created_at.utc.to_i / 1.hour.to_i
+
+    # Reuse previous group if it does not span too large an amount of time
+    previous_bucket = redis.get(redis_key).to_i
+    hour_bucket = previous_bucket if hour_bucket < previous_bucket + MAXIMUM_GROUP_SPAN_HOURS
+
+    # Do not track groups past a given inactivity time
+    # We do not concern ourselves with race conditions since we use hour buckets
+    redis.set(redis_key, hour_bucket, ex: MAXIMUM_GROUP_GAP_TIME)
+
+    "#{type_prefix}-#{hour_bucket}"
+  end
+
   def dismiss?
     DismissCondition.new(@notification).dismiss?
   end

config/application.rb

@@ -51,6 +51,8 @@
 require_relative '../lib/action_dispatch/remote_ip_extensions'
 require_relative '../lib/active_record/database_tasks_extensions'
 require_relative '../lib/active_record/batches'
+require_relative '../lib/active_record/with_recursive'
+require_relative '../lib/arel/union_parenthesizing'
 require_relative '../lib/simple_navigation/item_extensions'
 
 Bundler.require(:pam_authentication) if ENV['PAM_ENABLED'] == 'true'

config/routes/api.rb

@@ -327,6 +327,18 @@
     end
   end
 
+  namespace :v2_alpha do
+    resources :notifications, only: [:index, :show] do
+      collection do
+        post :clear
+      end
+
+      member do
+        post :dismiss
+      end
+    end
+  end
+
   namespace :web do
     resource :settings, only: [:update]
     resources :embeds, only: [:show]

db/migrate/20240513095755_add_group_key_to_notifications.rb

@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+class AddGroupKeyToNotifications < ActiveRecord::Migration[7.1]
+  def change
+    add_column :notifications, :group_key, :string
+  end
+end

db/migrate/20240513123807_add_index_notifications_on_account_id_and_group_key.rb

@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+class AddIndexNotificationsOnAccountIdAndGroupKey < ActiveRecord::Migration[7.1]
+  disable_ddl_transaction!
+
+  def change
+    add_index :notifications, [:account_id, :group_key], algorithm: :concurrently, where: 'group_key IS NOT NULL'
+  end
+end

db/schema.rb

@@ -724,6 +724,8 @@
     t.bigint "from_account_id", null: false
     t.string "type"
     t.boolean "filtered", default: false, null: false
+    t.string "group_key"
+    t.index ["account_id", "group_key"], name: "index_notifications_on_account_id_and_group_key", where: "(group_key IS NOT NULL)"
     t.index ["account_id", "id", "type"], name: "index_notifications_on_account_id_and_id_and_type", order: { id: :desc }
     t.index ["account_id", "id", "type"], name: "index_notifications_on_filtered", order: { id: :desc }, where: "(filtered = false)"
     t.index ["activity_id", "activity_type"], name: "index_notifications_on_activity_id_and_activity_type"