feat(products): catalog guards, dedup and batching (#5690)

## Context

Hardening on top of the catalog services (#6098) and GraphQL surface
(#6099): the immutability rules, the review fixes and the
list-performance work.

## Description

- Gate structural edits on attachment: a product, category or filter
priced by a plan or subscription cannot change identity or be deleted
(`attached_to_plan_or_subscription`); filter values freeze once a
subscription bills through the filter (`attached_to_subscriptions`).
- Creating a filter on a fixed product fails with
`not_allowed_for_product_type`.
- Reject duplicate filter value sets per product (`value_already_exist`
on `values`) — two identical slices would be unresolvable for the
billing engine; partial overlaps stay legal.
- Batch `filters_count`, `products_count` and
`attached_to_plan_or_subscription` through dataloader sources — one
grouped query per page instead of per row.

Author: rsempe

app/graphql/sources/attached_to_plan_or_subscription.rb

@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+module Sources
+  # Batches attached_to_plan_or_subscription? across catalog rows: one grouped
+  # query per applied-card side instead of two EXISTS per row.
+  #
+  # Keyed by the grouping the caller resolves through — a product id for
+  # products and their filters (a filter is attached when its product is), a
+  # product_category id for categories (attached through their products).
+  # Default scopes apply on both the applied cards and the joined models, so
+  # discarded rows stay excluded exactly as in the per-record checks.
+  #
+  # Usage in GraphQL types:
+  #   dataloader.with(Sources::AttachedToPlanOrSubscription, :product).load(object.id)
+  class AttachedToPlanOrSubscription < GraphQL::Dataloader::Source
+    GROUPINGS = {
+      product: [:rate_card, "rate_cards.product_id"],
+      product_category: [{rate_card: :product}, "products.product_category_id"]
+    }.freeze
+
+    def initialize(group_by)
+      @joins, @column = GROUPINGS.fetch(group_by)
+    end
+
+    def fetch(ids)
+      attached = PlanRateCard.joins(@joins).where(@column => ids).distinct.pluck(Arel.sql(@column)) |
+        SubscriptionRateCard.joins(@joins).where(@column => ids).distinct.pluck(Arel.sql(@column))
+
+      ids.map { attached.include?(it) }
+    end
+  end
+end

app/graphql/sources/count_by_foreign_key.rb

@@ -0,0 +1,27 @@
+# frozen_string_literal: true
+
+module Sources
+  # Batches per-row COUNT fields into a single grouped query.
+  #
+  # Prevents N+1 queries when a count is requested for several parent records
+  # in one GraphQL query (e.g., `products { filtersCount }`). The model's
+  # default scope applies, so soft-deleted rows stay excluded exactly as in
+  # the per-record `association.count`.
+  #
+  # Usage in GraphQL types:
+  #   dataloader.with(Sources::CountByForeignKey, ProductFilter, :product_id).load(object.id)
+  class CountByForeignKey < GraphQL::Dataloader::Source
+    def initialize(model, foreign_key)
+      @model = model
+      @foreign_key = foreign_key
+    end
+
+    def fetch(ids)
+      # reorder(nil) drops any default-scope ordering, which would otherwise
+      # leak into the GROUP BY and fail.
+      counts = @model.where(@foreign_key => ids).reorder(nil).group(@foreign_key).count
+
+      ids.map { counts.fetch(it, 0) }
+    end
+  end
+end

app/graphql/types/product_categories/object.rb

@@ -14,13 +14,18 @@ class Object < Types::BaseObject
       field :invoice_display_name, String, null: true
       field :name, String, null: false
 
+      field :attached_to_plan_or_subscription, Boolean, null: false
       field :products_count, Integer, null: false
 
       field :created_at, GraphQL::Types::ISO8601DateTime, null: false
       field :updated_at, GraphQL::Types::ISO8601DateTime, null: false
 
+      def attached_to_plan_or_subscription
+        dataloader.with(Sources::AttachedToPlanOrSubscription, :product_category).load(object.id)
+      end
+
       def products_count
-        object.products.count
+        dataloader.with(Sources::CountByForeignKey, Product, :product_category_id).load(object.id)
       end
     end
   end

app/graphql/types/product_categories/update_input.rb

@@ -7,6 +7,7 @@ class UpdateInput < BaseInputObject
 
       argument :id, ID, required: true
 
+      argument :code, String, required: false
       argument :description, String, required: false
       argument :invoice_display_name, String, required: false
       argument :name, String, required: false

app/graphql/types/product_filters/object.rb

@@ -11,6 +11,7 @@ class Object < Types::BaseObject
       field :id, ID, null: false
       field :organization, Types::Organizations::OrganizationType
 
+      field :attached_to_plan_or_subscription, Boolean, null: false
       field :code, String, null: false
       field :description, String, null: true
       field :invoice_display_name, String, null: true
@@ -21,6 +22,12 @@ class Object < Types::BaseObject
 
       field :created_at, GraphQL::Types::ISO8601DateTime, null: false
       field :updated_at, GraphQL::Types::ISO8601DateTime, null: false
+
+      # A filter is attached when its product is — resolved by product_id so
+      # the batch is shared with the products' own attachment checks.
+      def attached_to_plan_or_subscription
+        dataloader.with(Sources::AttachedToPlanOrSubscription, :product).load(object.product_id)
+      end
     end
   end
 end

app/graphql/types/product_filters/update_input.rb

@@ -7,6 +7,7 @@ class UpdateInput < BaseInputObject
 
       argument :id, ID, required: true
 
+      argument :code, String, required: false
       argument :description, String, required: false
       argument :invoice_display_name, String, required: false
       argument :name, String, required: false

app/graphql/types/products/object.rb

@@ -11,6 +11,7 @@ class Object < Types::BaseObject
       field :id, ID, null: false
       field :organization, Types::Organizations::OrganizationType
 
+      field :attached_to_plan_or_subscription, Boolean, null: false
       field :code, String, null: false
       field :description, String, null: true
       field :invoice_display_name, String, null: true
@@ -25,8 +26,12 @@ class Object < Types::BaseObject
       field :created_at, GraphQL::Types::ISO8601DateTime, null: false
       field :updated_at, GraphQL::Types::ISO8601DateTime, null: false
 
+      def attached_to_plan_or_subscription
+        dataloader.with(Sources::AttachedToPlanOrSubscription, :product).load(object.id)
+      end
+
       def filters_count
-        object.filters.count
+        dataloader.with(Sources::CountByForeignKey, ProductFilter, :product_id).load(object.id)
       end
     end
   end

app/graphql/types/products/update_input.rb

@@ -7,9 +7,11 @@ class UpdateInput < BaseInputObject
 
       argument :id, ID, required: true
 
+      argument :code, String, required: false
       argument :description, String, required: false
       argument :invoice_display_name, String, required: false
       argument :name, String, required: false
+      argument :product_category_id, ID, required: false
     end
   end
 end

app/models/product_filter.rb

@@ -11,9 +11,14 @@ class ProductFilter < ApplicationRecord
 
   has_many :values, class_name: "ProductFilterValue"
   has_many :billable_metric_filters, through: :values
+  has_many :rate_cards
 
   delegate :attached_to_plan_or_subscription?, to: :product
 
+  def attached_to_subscriptions?
+    SubscriptionRateCard.joins(:rate_card).where(rate_cards: {product_filter_id: id}).exists?
+  end
+
   validates :name, presence: true
   validates :code,
     presence: true,

app/services/product_categories/destroy_service.rb

@@ -17,6 +17,10 @@ def initialize(product_category:)
     def call
       return result.not_found_failure!(resource: "product_category") unless product_category
 
+      if product_category.attached_to_plan_or_subscription?
+        return result.single_validation_failure!(field: :product_category, error_code: "attached_to_plan_or_subscription")
+      end
+
       ActiveRecord::Base.transaction do
         product_category.products.find_each do |product|
           Products::DestroyService.call!(product:)