Manual Audience Segments API write endpoints (#19476)

* add composite index to segmented_users table

* bulk queries for segmented users

* create segment endpoint

* adding and removing users

* deleting segments 😪

* fix time precision in specs

* api docs 🎉

* pluck improvements that couldn't be cherry-picked

* better service objects; also introduce batch upserting

* docs???

app/controllers/api/v1/audience_segments_controller.rb

@@ -0,0 +1,58 @@
+module Api
+  module V1
+    class AudienceSegmentsController < ApiController
+      MAX_USER_IDS = 10_000
+
+      before_action :authenticate_with_api_key!
+      before_action :require_admin
+      before_action :restrict_user_ids, only: %i[add_users remove_users]
+
+      def create
+        @segment = AudienceSegment.create!(type_of: "manual")
+
+        render json: @segment, status: :created
+      end
+
+      def destroy
+        @segment = scope.find(params[:id])
+
+        if DisplayAd.where(audience_segment_id: @segment.id).any?
+          render json: { error: "Segments cannot be deleted while in use by any billboards" }, status: :conflict
+        else
+          @segment.segmented_users.in_batches.delete_all
+          result = @segment.destroy
+          render json: @segment, status: (result ? :ok : :conflict)
+        end
+      end
+
+      def add_users
+        @segment = scope.find(params[:id])
+
+        render json: SegmentedUsers::BulkUpsert.call(@segment, user_ids: @user_ids)
+      end
+
+      def remove_users
+        @segment = scope.find(params[:id])
+
+        render json: SegmentedUsers::BulkDelete.call(@segment, user_ids: @user_ids)
+      end
+
+      private
+
+      def require_admin
+        authorize AudienceSegment, :access?, policy_class: InternalPolicy
+      end
+
+      def restrict_user_ids
+        @user_ids = params.permit(user_ids: []).require(:user_ids)
+        return unless @user_ids.size > MAX_USER_IDS
+
+        render json: { error: "Too many user IDs provided" }, status: :unprocessable_entity
+      end
+
+      def scope
+        AudienceSegment.manual
+      end
+    end
+  end
+end

app/helpers/notifications_helper.rb

@@ -34,7 +34,7 @@ module NotificationsHelper
       )
     else
       I18n.t(action, user: key_to_link.call("user"))
-    end.html_safe # rubocop:disable Rails/OutputSafety
+    end.html_safe
   end
 
   def mod_comment_user(data)

app/models/notification.rb

@@ -103,8 +103,8 @@ class Notification < ApplicationRecord
     def send_moderation_notification(notifiable)
       return unless [Comment, Article].include?(notifiable.class)
 
-      if notifiable.instance_of?(Comment)
-        return if UserBlock.blocking?(notifiable.commentable.user_id, notifiable.user_id)
+      if notifiable.instance_of?(Comment) && UserBlock.blocking?(notifiable.commentable.user_id, notifiable.user_id)
+        return
       end
 
       Notifications::CreateRoundRobinModerationNotificationsWorker.perform_async(notifiable.id, notifiable.class.to_s)

app/services/segmented_users/bulk_delete.rb

@@ -0,0 +1,35 @@
+module SegmentedUsers
+  # Preferred way to quickly remove a large number of users from an AudienceSegment
+  class BulkDelete
+    Result = Struct.new(:succeeded, :failed, keyword_init: true)
+
+    def self.call(audience_segment, user_ids:)
+      new(audience_segment).call(user_ids)
+    end
+
+    # @param audience_segment [AudienceSegment] the segment to remove users from
+    def initialize(audience_segment)
+      @audience_segment = audience_segment
+    end
+
+    # Deletes the provided users from the AudienceSegment in batches.
+    # It touches the segment if any users were successfully deleted.
+    #
+    # Warning: the joining `SegmentedUsers` records are deleted without triggering any
+    # application-defined callbacks. Doing so is the responsibility of the caller.
+    #
+    # @param user_ids [Array<Integer>] a list of `User` ids to process
+    # @return [SegmentedUsers::BulkDelete::Result]
+    def call(user_ids)
+      return unless @audience_segment.persisted?
+
+      segmented_users = @audience_segment.segmented_users.where(user_id: user_ids)
+      valid_user_ids = segmented_users.pluck(:user_id)
+      deleted_count = segmented_users.in_batches.delete_all
+
+      @audience_segment.touch if deleted_count.positive?
+
+      Result.new(succeeded: valid_user_ids, failed: user_ids - valid_user_ids)
+    end
+  end
+end

app/services/segmented_users/bulk_upsert.rb

@@ -0,0 +1,80 @@
+module SegmentedUsers
+  # Preferred way to add a large number of users to an AudienceSegment
+  # Normal ActiveRecord usage emits one INSERT statement per row.
+  class BulkUpsert
+    Result = Struct.new(:succeeded, :failed, keyword_init: true)
+
+    def self.call(audience_segment, user_ids:)
+      new(audience_segment).call(user_ids)
+    end
+
+    # @param audience_segment [AudienceSegment] the segment to add users to
+    def initialize(audience_segment)
+      @audience_segment = audience_segment
+    end
+
+    # Upserts the provided users into the AudienceSegment in batches.
+    # It touches the `SegmentedUser` record of any users already in the list, as
+    # well as the segment itself if any users were successfully upserted.
+    #
+    # Warning: the joining `SegmentedUsers` records are created without triggering any
+    # application-defined callbacks. Doing so is the responsibility of the caller.
+    #
+    # @param user_ids [Array<Integer>] a list of `User` ids to process
+    # @return [SegmentedUsers::BulkUpsert::Result]
+    def call(user_ids)
+      return unless audience_segment.persisted?
+
+      @upsert_time = Time.current
+
+      valid_user_ids = User.where(id: user_ids).ids
+      upserted_user_ids = upsert_in_batches(valid_user_ids)
+
+      audience_segment.touch unless upserted_user_ids.empty?
+
+      Result.new(succeeded: upserted_user_ids, failed: user_ids - upserted_user_ids)
+    end
+
+    private
+
+    attr_reader :audience_segment, :upsert_time
+
+    def upsert_in_batches(user_ids)
+      result = []
+
+      user_ids.in_groups_of(1000, false) do |ids|
+        succeeded = perform_upsert(ids).rows.flatten
+        result.concat(succeeded)
+      end
+
+      result
+    end
+
+    def perform_upsert(user_ids_batch)
+      segmented_users = build_records(user_ids_batch)
+
+      # We only want to touch the `updated_at` column of records that already exist,
+      # but specifying that causes ActiveRecord to emit malformed SQL (multiple assignments to the same column).
+      # Turning off Rails' automatic timestamp management via `record_timestamps`
+      # and managing timestamps ourselves yields the correct query.
+      SegmentedUser.upsert_all(
+        segmented_users,
+        unique_by: :index_segmented_users_on_audience_segment_and_user,
+        update_only: [:updated_at],
+        returning: ["user_id"],
+        record_timestamps: false,
+      )
+    end
+
+    def build_records(user_ids_batch)
+      user_ids_batch.map do |user_id|
+        {
+          audience_segment_id: audience_segment.id,
+          user_id: user_id,
+          created_at: upsert_time,
+          updated_at: upsert_time
+        }
+      end
+    end
+  end
+end

config/routes.rb

@@ -55,6 +55,11 @@ Rails.application.routes.draw do
           put "unpublish", on: :member
         end
 
+        resources :segments, controller: "audience_segments", only: %i[create destroy] do
+          put "add_users", on: :member
+          put "remove_users", on: :member
+        end
+
         resources :pages, only: %i[index show create update destroy]
 
         draw :api

db/migrate/20230511194704_add_composite_index_to_segmented_users.rb

@@ -0,0 +1,11 @@
+class AddCompositeIndexToSegmentedUsers < ActiveRecord::Migration[7.0]
+  disable_ddl_transaction!
+
+  def change
+    add_index :segmented_users,
+              %i[audience_segment_id user_id],
+              name: "index_segmented_users_on_audience_segment_and_user",
+              unique: true,
+              algorithm: :concurrently
+  end
+end

db/schema.rb

@@ -998,6 +998,7 @@ ActiveRecord::Schema[7.0].define(version: 2023_05_17_132219) do
     t.datetime "created_at", null: false
     t.datetime "updated_at", null: false
     t.bigint "user_id", null: false
+    t.index ["audience_segment_id", "user_id"], name: "index_segmented_users_on_audience_segment_and_user", unique: true
     t.index ["audience_segment_id"], name: "index_segmented_users_on_audience_segment_id"
     t.index ["user_id"], name: "index_segmented_users_on_user_id"
   end

spec/requests/api/v1/audience_segments_spec.rb

@@ -0,0 +1,218 @@
+require "rails_helper"
+
+RSpec.describe "Api::V1::AudienceSegments" do
+  let(:v1_headers) { { "content-type" => "application/json", "Accept" => "application/vnd.forem.api-v1+json" } }
+
+  let(:api_secret) { create(:api_secret) }
+  let(:admin) { api_secret.user }
+  let(:headers) { v1_headers.merge({ "api-key" => api_secret.secret }) }
+
+  before do
+    admin.add_role(:admin)
+  end
+
+  shared_examples "an admin-only protected resource" do
+    context "when no API secret is provided" do
+      let(:headers) { v1_headers }
+
+      it "returns unauthorized" do
+        make_request
+        expect(response).to have_http_status(:unauthorized)
+      end
+    end
+
+    context "when the authenticated user is not an admin" do
+      let(:regular_api_secret) { create(:api_secret) }
+      let(:headers) { v1_headers.merge({ "api-key" => regular_api_secret.secret }) }
+
+      it "returns unauthorized" do
+        make_request
+        expect(response).to have_http_status(:unauthorized)
+      end
+    end
+  end
+
+  shared_examples "an endpoint for only manual audience segments" do
+    it "returns not found if the segment is automatic" do
+      segment.update!(type_of: "trusted")
+      make_request
+      expect(response).to have_http_status(:not_found)
+    end
+
+    it "returns not found if the segment has been deleted" do
+      segment.destroy
+      make_request
+      expect(response).to have_http_status(:not_found)
+    end
+  end
+
+  describe "POST /api/segments" do
+    subject(:make_request) { post api_segments_path, headers: headers }
+
+    it_behaves_like "an admin-only protected resource"
+
+    it "creates a new manual audience segment" do
+      make_request
+
+      expect(response).to have_http_status(:created)
+      expect(response.media_type).to eq("application/json")
+
+      segment = AudienceSegment.last
+      expect(segment.manual?).to be(true)
+      expect(response.parsed_body).to include(
+        "id" => segment.id,
+        "type_of" => "manual",
+      )
+    end
+  end
+
+  describe "PUT /api/segments/:id/add_users" do
+    subject(:make_request) do
+      put add_users_api_segment_path(segment.id), params: { user_ids: user_ids }, headers: headers, as: :json
+    end
+
+    let(:segment) { AudienceSegment.create!(type_of: "manual") }
+    let(:users) { create_list(:user, 3) }
+    let(:user_ids) { users.map(&:id) }
+
+    it_behaves_like "an admin-only protected resource"
+
+    it_behaves_like "an endpoint for only manual audience segments"
+
+    it "adds the provided users" do
+      make_request
+
+      expect(response).to have_http_status(:success)
+      expect(response.media_type).to eq("application/json")
+
+      expect(segment.users).to match_array(users)
+
+      data = response.parsed_body
+      expect(data["succeeded"]).to match_array(user_ids)
+      expect(data["failed"]).to be_empty
+    end
+
+    it "returns user ids that failed to be added" do
+      fake_user_ids = [999_999, 777_777]
+      params = { user_ids: user_ids + fake_user_ids }
+      put add_users_api_segment_path(segment.id), params: params, headers: headers, as: :json
+
+      expect(response).to have_http_status(:success)
+      expect(response.media_type).to eq("application/json")
+
+      expect(segment.users).to match_array(users)
+
+      data = response.parsed_body
+      expect(data["succeeded"]).to match_array(user_ids)
+      expect(data["failed"]).to match_array(fake_user_ids)
+    end
+
+    it "handles empty, missing or too large user_ids" do
+      put add_users_api_segment_path(segment.id), headers: headers
+      expect(response).to have_http_status(:unprocessable_entity)
+
+      put add_users_api_segment_path(segment.id), params: { user_ids: [] }, headers: headers, as: :json
+      expect(response).to have_http_status(:unprocessable_entity)
+
+      params = { user_ids: (1..10_100).to_a }
+      put add_users_api_segment_path(segment.id), params: params, headers: headers, as: :json
+      expect(response).to have_http_status(:unprocessable_entity)
+    end
+  end
+
+  describe "PUT /api/segments/:id/remove_users" do
+    subject(:make_request) do
+      put remove_users_api_segment_path(segment.id), params: { user_ids: user_ids }, headers: headers, as: :json
+    end
+
+    let(:segment) { AudienceSegment.create!(type_of: "manual") }
+    let(:users) { create_list(:user, 3) }
+    let(:user_ids) { users.map(&:id) }
+    let(:retained_users) { create_list(:user, 3) }
+    let(:users_not_in_segment) { create_list(:user, 3) }
+
+    before do
+      segment.users << users
+      segment.users << retained_users
+    end
+
+    it_behaves_like "an admin-only protected resource"
+
+    it_behaves_like "an endpoint for only manual audience segments"
+
+    it "removes only the provided users if they are part of the segment" do
+      make_request
+
+      expect(response).to have_http_status(:success)
+      expect(response.media_type).to eq("application/json")
+
+      expect(segment.users).to match_array(retained_users)
+
+      data = response.parsed_body
+      expect(data["succeeded"]).to match_array(user_ids)
+      expect(data["failed"]).to be_empty
+    end
+
+    it "returns user ids that failed to be removed" do
+      ids_to_fail = [*users_not_in_segment.map(&:id), 123_456]
+      params = { user_ids: user_ids + ids_to_fail }
+      put remove_users_api_segment_path(segment.id), params: params, headers: headers, as: :json
+
+      expect(response).to have_http_status(:success)
+      expect(response.media_type).to eq("application/json")
+
+      expect(segment.users).to match_array(retained_users)
+
+      data = response.parsed_body
+      expect(data["succeeded"]).to match_array(user_ids)
+      expect(data["failed"]).to match_array(ids_to_fail)
+    end
+
+    it "handles empty, missing or too large user_ids" do
+      put remove_users_api_segment_path(segment.id), headers: headers
+      expect(response).to have_http_status(:unprocessable_entity)
+
+      put remove_users_api_segment_path(segment.id), params: { user_ids: [] }, headers: headers, as: :json
+      expect(response).to have_http_status(:unprocessable_entity)
+
+      params = { user_ids: (1..10_100).to_a }
+      put remove_users_api_segment_path(segment.id), params: params, headers: headers, as: :json
+      expect(response).to have_http_status(:unprocessable_entity)
+    end
+  end
+
+  describe "DELETE /api/segments/:id" do
+    subject(:make_request) { delete api_segment_path(segment.id), headers: headers }
+
+    let(:segment) { AudienceSegment.create!(type_of: "manual") }
+    let(:users) { create_list(:user, 3) }
+    let(:billboard) { create(:display_ad, published: true, approved: true, type_of: "community") }
+
+    it_behaves_like "an admin-only protected resource"
+
+    it_behaves_like "an endpoint for only manual audience segments"
+
+    it "destroys the segment and cleans up its list of users" do
+      segment.users << users
+
+      make_request
+
+      expect(response).to have_http_status(:success)
+      expect(response.media_type).to eq("application/json")
+      expect(AudienceSegment.exists?(segment.id)).to be(false)
+      expect(SegmentedUser.where(audience_segment_id: segment.id)).to be_empty
+    end
+
+    it "does not destroy the segment or its user list if it is associated with a billboard" do
+      segment.users << users
+      billboard.update!(audience_segment: segment)
+
+      make_request
+
+      expect(response).to have_http_status(:conflict)
+      expect(response.media_type).to eq("application/json")
+      expect(response.parsed_body.keys).to include("error")
+      expect(segment.reload.users).to match_array(users)
+    end
+  end
+end

spec/requests/api/v1/docs/audience_segments_spec.rb

@@ -0,0 +1,289 @@
+require "rails_helper"
+require "swagger_helper"
+
+def id_schema
+  {
+    type: :integer,
+    format: :int32,
+    minimum: 1
+  }
+end
+
+# rubocop:disable RSpec/VariableName
+# rubocop:disable RSpec/EmptyExampleGroup
+# rubocop:disable Layout/LineLength
+
+RSpec.describe "Api::V1::Docs::AudienceSegments" do
+  let(:Accept) { "application/vnd.forem.api-v1+json" }
+  let(:admin_api_secret) { create(:api_secret) }
+  let(:regular_api_secret) { create(:api_secret) }
+  let(:segment) { AudienceSegment.create!(type_of: "manual") }
+  let(:automatic_segment) { AudienceSegment.create!(type_of: "trusted") }
+  let(:users) { create_list(:user, 3) }
+
+  before do
+    admin_api_secret.user.add_role(:admin)
+  end
+
+  describe "POST /segments" do
+    path "/api/segments" do
+      post "Create a manually managed audience segment" do
+        tags "segments"
+        description "This endpoint allows the client to create a new audience segment.\n\nAn audience segment is a group of users that can be targeted by a DisplayAd/Billboard/Widget. This API only permits managing segments you create and maintain yourself."
+        operationId "createSegment"
+        produces "application/json"
+        consumes "application/json"
+
+        response "201", "A manually managed audience segment" do
+          let(:"api-key") { admin_api_secret.secret }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "401", "Unauthorized" do
+          let(:"api-key") { nil }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "401", "Unauthorized" do
+          let(:"api-key") { regular_api_secret.secret }
+
+          add_examples
+
+          run_test!
+        end
+      end
+    end
+  end
+
+  describe "DELETE /segments/:id" do
+    path "/api/segments/{id}" do
+      delete "Delete a manually managed audience segment" do
+        tags "segments"
+        description "This endpoint allows the client to delete an audience segment specified by ID.\n\nAudience segments cannot be deleted if there are still any DisplayAds/Billboards/Widgets using them."
+        operationId "deleteSegment"
+        produces "application/json"
+        consumes "application/json"
+
+        parameter name: :id, in: :path, required: true, schema: id_schema
+
+        response "200", "The deleted audience segment" do
+          let(:"api-key") { admin_api_secret.secret }
+          let(:id) { segment.id }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "401", "Unauthorized" do
+          let(:"api-key") { nil }
+          let(:id) { segment.id }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "401", "Unauthorized" do
+          let(:"api-key") { regular_api_secret.secret }
+          let(:id) { segment.id }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "404", "Audience Segment Not Found" do
+          let(:"api-key") { admin_api_secret.secret }
+          let(:id) { automatic_segment.id }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "409", "Audience segment could not be deleted" do
+          let(:"api-key") { admin_api_secret.secret }
+          let(:id) { segment.id }
+          let(:billboard) { create(:display_ad, published: true, approved: true) }
+
+          before do
+            billboard.update!(audience_segment: segment)
+          end
+
+          add_examples
+
+          run_test!
+        end
+      end
+    end
+  end
+
+  describe "PUT /segments/:id/add_users" do
+    path "/api/segments/{id}/add_users" do
+      put "Add users to a manually managed audience segment" do
+        tags "segments"
+        description "This endpoint allows the client to add users in bulk to an audience segment specified by ID.\n\nSuccesses are users that were included in the segment (even if they were already in it), and failures are users that could not be added to the segment."
+        operationId "addUsersToSegment"
+        produces "application/json"
+        consumes "application/json"
+
+        parameter name: :id, in: :path, required: true, schema: id_schema
+        parameter name: :user_ids,
+                  in: :body,
+                  schema: { "$ref": "#/components/schemas/SegmentUserIds" }
+
+        response "200", "Result of adding the users to the segment." do
+          let(:"api-key") { admin_api_secret.secret }
+          let(:id) { segment.id }
+          let(:user_ids) { { user_ids: users.map(&:id) } }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "401", "Unauthorized" do
+          let(:"api-key") { nil }
+          let(:id) { segment.id }
+          let(:user_ids) { { user_ids: users.map(&:id) } }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "401", "Unauthorized" do
+          let(:"api-key") { regular_api_secret.secret }
+          let(:id) { segment.id }
+          let(:user_ids) { { user_ids: users.map(&:id) } }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "404", "Audience Segment Not Found" do
+          let(:"api-key") { admin_api_secret.secret }
+          let(:id) { automatic_segment.id }
+          let(:user_ids) { { user_ids: users.map(&:id) } }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "422", "Unprocessable Entity" do
+          let(:"api-key") { admin_api_secret.secret }
+          let(:id) { segment.id }
+          let(:user_ids) { { user_ids: [] } }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "422", "Unprocessable Entity" do
+          let(:"api-key") { admin_api_secret.secret }
+          let(:id) { segment.id }
+          let(:user_ids) { (1..10_100).to_a }
+
+          add_examples
+
+          run_test!
+        end
+      end
+    end
+  end
+
+  describe "PUT /segments/:id/remove_users" do
+    path "/api/segments/{id}/remove_users" do
+      put "Remove users from a manually managed audience segment" do
+        tags "segments"
+        description "This endpoint allows the client to remove users in bulk from an audience segment specified by ID.\n\nSuccesses are users that were removed; failures are users that weren't a part of the segment."
+        operationId "removeUsersFromSegment"
+        produces "application/json"
+        consumes "application/json"
+
+        parameter name: :id, in: :path, required: true, schema: id_schema
+        parameter name: :user_ids,
+                  in: :body,
+                  schema: { "$ref": "#/components/schemas/SegmentUserIds" }
+
+        before do
+          segment.users << users
+        end
+
+        response "200", "Result of removing the users to the segment." do
+          let(:"api-key") { admin_api_secret.secret }
+          let(:id) { segment.id }
+          let(:user_ids) { { user_ids: users.map(&:id) } }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "401", "Unauthorized" do
+          let(:"api-key") { nil }
+          let(:id) { segment.id }
+          let(:user_ids) { { user_ids: users.map(&:id) } }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "401", "Unauthorized" do
+          let(:"api-key") { regular_api_secret.secret }
+          let(:id) { segment.id }
+          let(:user_ids) { { user_ids: users.map(&:id) } }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "404", "Audience Segment Not Found" do
+          let(:"api-key") { admin_api_secret.secret }
+          let(:id) { automatic_segment.id }
+          let(:user_ids) { { user_ids: users.map(&:id) } }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "422", "Unprocessable Entity" do
+          let(:"api-key") { admin_api_secret.secret }
+          let(:id) { segment.id }
+          let(:user_ids) { { user_ids: [] } }
+
+          add_examples
+
+          run_test!
+        end
+
+        response "422", "Unprocessable Entity" do
+          let(:"api-key") { admin_api_secret.secret }
+          let(:id) { segment.id }
+          let(:user_ids) { (1..10_100).to_a }
+
+          add_examples
+
+          run_test!
+        end
+      end
+    end
+  end
+end
+
+# rubocop:enable RSpec/VariableName
+# rubocop:enable RSpec/EmptyExampleGroup
+# rubocop:enable Layout/LineLength

spec/services/segmented_users/bulk_delete_spec.rb

@@ -0,0 +1,32 @@
+require "rails_helper"
+
+RSpec.describe SegmentedUsers::BulkDelete, type: :service do
+  let(:audience_segment) { AudienceSegment.create!(type_of: "manual") }
+  let(:users) { create_list(:user, 5) }
+
+  it "removes only the passed users from the segment" do
+    retained_users = create_list(:user, 3)
+
+    audience_segment.users << users
+    audience_segment.users << retained_users
+
+    user_ids = users.map(&:id)
+    result = described_class.call(audience_segment, user_ids: user_ids)
+    expect(result.succeeded).to match_array(user_ids)
+    expect(result.failed).to be_empty
+    expect(audience_segment.users).to match_array(retained_users)
+  end
+
+  it "gracefully handles users that don't exist or aren't part of the segment" do
+    user_in_segment = users.first
+    audience_segment.users << user_in_segment
+
+    user_not_in_segment = create(:user)
+    user_ids = [user_in_segment.id, user_not_in_segment.id, "foo", 1_234_567_890]
+
+    result = described_class.call(audience_segment, user_ids: user_ids)
+    expect(result.succeeded).to contain_exactly(user_in_segment.id)
+    expect(result.failed).to contain_exactly(user_not_in_segment.id, "foo", 1_234_567_890)
+    expect(audience_segment.reload.users).to be_empty
+  end
+end

spec/services/segmented_users/bulk_upsert_spec.rb

@@ -0,0 +1,86 @@
+require "rails_helper"
+
+RSpec.describe SegmentedUsers::BulkUpsert, type: :service do
+  let(:audience_segment) { AudienceSegment.create!(type_of: "manual") }
+  let(:users) { create_list(:user, 5) }
+
+  it "adds the users to the segment" do
+    user_ids = users.map(&:id)
+    result = described_class.call(audience_segment, user_ids: user_ids)
+    expect(result.succeeded).to match_array(user_ids)
+    expect(result.failed).to be_empty
+    expect(audience_segment.users).to match_array(users)
+  end
+
+  it "retains any existing users in the segment" do
+    existing_user = create(:user)
+    audience_segment.users << [existing_user]
+
+    user_ids = users.map(&:id)
+    result = described_class.call(audience_segment, user_ids: user_ids)
+    expect(result.succeeded).to match_array(user_ids)
+    expect(result.failed).to be_empty
+    expect(audience_segment.users).to contain_exactly(*users, existing_user)
+  end
+
+  it "only touches segmented users already in the list" do
+    created_time = Time.current
+    upsert_time = 3.days.from_now
+    existing_users = create_list(:user, 3)
+    existing_user_ids = existing_users.map(&:id)
+
+    Timecop.freeze(created_time) do
+      audience_segment.users << existing_users
+    end
+
+    Timecop.freeze(upsert_time) do
+      user_ids = users.map(&:id) + existing_user_ids
+      result = described_class.call(audience_segment, user_ids: user_ids)
+      expect(result.succeeded).to match_array(user_ids)
+      expect(result.failed).to be_empty
+
+      existing_segmented_users = audience_segment.segmented_users.where(user_id: existing_user_ids)
+      expect(existing_segmented_users.map(&:created_at)).to all(be_within(1.second).of(created_time))
+      expect(existing_segmented_users.map(&:updated_at)).to all(be_within(1.second).of(upsert_time))
+    end
+  end
+
+  it "only touches the segment if any users were successfully upserted" do
+    user = create(:user)
+    time = Time.current
+
+    Timecop.freeze(time) do
+      described_class.call(audience_segment, user_ids: [user.id])
+      expect(audience_segment.updated_at).to be_within(1.second).of(time)
+
+      Timecop.travel(1.day.from_now) do
+        described_class.call(audience_segment, user_ids: [])
+        expect(audience_segment.updated_at).to be_within(1.second).of(time)
+      end
+
+      Timecop.travel(2.days.from_now) do
+        described_class.call(audience_segment, user_ids: [user.id])
+        expect(audience_segment.updated_at).to be_within(1.second).of(Time.current)
+      end
+    end
+  end
+
+  it "gracefully handles users that don't exist" do
+    real_user = create(:user)
+    user_ids = [real_user.id, "foo", 1_234_567_890]
+    result = described_class.call(audience_segment, user_ids: user_ids)
+
+    expect(result.succeeded).to contain_exactly(real_user.id)
+    expect(result.failed).to contain_exactly("foo", 1_234_567_890)
+    expect(audience_segment.users).to contain_exactly(real_user)
+  end
+
+  it "returns immediately if the segment is not persisted" do
+    audience_segment.destroy
+
+    user_ids = users.map(&:id)
+    result = described_class.call(audience_segment, user_ids: user_ids)
+    expect(result).to be_nil
+    expect(SegmentedUser.where(audience_segment_id: audience_segment.id)).to be_empty
+  end
+end

spec/swagger_helper.rb

@@ -430,6 +430,25 @@ The default maximum value can be overridden by \"API_PER_PAGE_MAX\" environment
                 }
             },
             required: %w[name body_markdown placement_area]
+          },
+          Segment: {
+            description: "A manually managed audience segment",
+            type: "object",
+            properties: {
+              id: { type: :integer, description: "The ID of the segment" },
+              type_of: { type: :string, enum: ["manual"], default: "manual", description: "Marks the segment as manually managed (other types are internal)" },
+              user_count: { type: :integer, description: "The current number of users in the segment" }
+            }
+          },
+          SegmentUserIds: {
+            type: "object",
+            properties: {
+              user_ids: {
+                type: :array,
+                items: { type: :integer },
+                maxItems: 10_000
+              }
+            }
           }
         }
       }