class AddUniqueIndexToInventorySnapshots < ActiveRecord::Migration[7.1]
disable_ddl_transaction!
def change
add_index :inventory_snapshots,
[:warehouse_id, :sku],
unique: true,
algorithm: :concurrently,
name: "index_inventory_snapshots_on_warehouse_and_sku"
end
end
class InventorySnapshot < ApplicationRecord
belongs_to :warehouse
validates :sku, presence: true
validates :quantity, numericality: { greater_than_or_equal_to: 0 }
validates :sku, uniqueness: { scope: :warehouse_id }
def self.upsert_keys
%i[warehouse_id sku]
end
def self.mutable_columns
%i[quantity updated_at]
end
end
class InventorySyncService
def initialize(warehouse)
@warehouse = warehouse
end
def sync!(payloads)
rows = dedupe_rows(payloads.map { |p| build_row(p) })
return { upserted: 0 } if rows.empty?
result = InventorySnapshot.upsert_all(
rows,
unique_by: :index_inventory_snapshots_on_warehouse_and_sku,
update_only: InventorySnapshot.mutable_columns,
returning: %i[id sku]
)
{ upserted: result.length }
end
private
def build_row(payload)
now = Time.current
{
warehouse_id: @warehouse.id,
sku: payload.fetch(:sku).to_s.strip.upcase,
quantity: Integer(payload.fetch(:quantity)),
created_at: now,
updated_at: now
}
end
def dedupe_rows(rows)
# Postgres rejects two rows hitting the same conflict target in one statement;
# keep the last occurrence per [warehouse_id, sku].
rows.index_by { |r| [r[:warehouse_id], r[:sku]] }.values
end
end
This snippet shows how a Rails service ingests batches of external records efficiently and idempotently using upsert_all backed by a composite unique index. The pattern is common when syncing data from a third-party API or webhook stream, where the same logical record may arrive many times and one-row-at-a-time save! calls would be far too slow.
The AddUniqueIndexToInventorySnapshots migration establishes the invariant the whole design depends on. It adds a unique index on [:warehouse_id, :sku], which gives Postgres a target for its ON CONFLICT clause. Without a matching unique constraint, upsert_all has nothing to detect a conflict against and would simply insert duplicates. The migration is disable_ddl_transaction! and uses algorithm: :concurrently so the index build does not take a heavy lock on a large, live table.
The InventorySnapshot model declares upsert_keys as a class method naming the conflict columns, keeping the migration and the service in agreement about what "the same record" means. It also carries a normal validates uniqueness rule for the single-record code paths; note that upsert_all bypasses validations and callbacks entirely, so the database index is the real source of truth, not the model.
The InventorySyncService is where the bulk write happens. sync! maps raw payloads into plain attribute hashes, stamps created_at/updated_at manually (since callbacks are skipped), and calls upsert_all with unique_by: pointing at the composite index and update_only: listing the mutable columns. The update_only option means conflicting rows only touch quantity and updated_at, leaving other columns untouched. dedupe_rows collapses duplicates within a single batch by [warehouse_id, sku], because Postgres rejects a statement that tries to affect the same conflict target twice in one command — a subtle but frequent production error.
The trade-offs are worth understanding: upsert_all is dramatically faster and atomic per batch, but it silently skips validations, callbacks, and updated_at auto-stamping, so any invariant that matters must live in the schema. This approach fits high-volume ingestion where correctness is enforced by the database and business callbacks are not required.
Related snips
class CommentsController < ApplicationController
before_action :set_post
def create
@comment = @post.comments.build(comment_params)System test: asserting Turbo Stream responses
class Post < ApplicationRecord
belongs_to :author, class_name: 'User'
has_many :comments, dependent: :destroy
scope :published, -> { where.not(published_at: nil).where('published_at <= ?', Time.current) }
scope :draft, -> { where(published_at: nil) }ActiveRecord scopes for reusable query logic
timestamp = request.headers.fetch('X-Signature-Timestamp')
signature = request.headers.fetch('X-Signature')
payload = request.raw_post
data = "#{timestamp}.#{payload}"
expected = OpenSSL::HMAC.hexdigest('SHA256', ENV.fetch('WEBHOOK_SECRET'), data)HMAC signed API requests for webhook and partner integrity
module Api
module V1
class UsersController < BaseController
def show
user = User.includes(:profile).find(params[:id])ETags for conditional requests and caching
class PostsController < ApplicationController
def index
@posts = Post.includes(:author)
.order(created_at: :desc)
.page(params[:page])
.per(10)Turbo Frames: infinite scroll with lazy-loading frame
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
Share this code
Here's the card — post it anywhere.
