class InventoryLevel < ApplicationRecord
belongs_to :warehouse
UPSERT_COLUMNS = %w[warehouse_id sku on_hand reserved updated_at].freeze
def self.upsert_counts(rows)
return if rows.blank?
payload = rows.map do |row|
row.stringify_keys.slice(*UPSERT_COLUMNS)
end
upsert_all(
payload,
unique_by: :index_inventory_levels_on_warehouse_and_sku,
returning: %w[sku on_hand]
)
end
def available
on_hand - reserved
end
end
CREATE TABLE inventory_levels (
id BIGSERIAL PRIMARY KEY,
warehouse_id BIGINT NOT NULL REFERENCES warehouses(id),
sku VARCHAR(64) NOT NULL,
on_hand INTEGER NOT NULL DEFAULT 0,
reserved INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
CONSTRAINT on_hand_non_negative CHECK (on_hand >= 0),
CONSTRAINT reserved_non_negative CHECK (reserved >= 0)
);
CREATE UNIQUE INDEX index_inventory_levels_on_warehouse_and_sku
ON inventory_levels (warehouse_id, sku);
class ReconcileInventoryJob
include Sidekiq::Job
sidekiq_options queue: :inventory, retry: 5
BATCH_SIZE = 2_000
def perform(warehouse_id, feed_rows)
now = Time.current
rows = feed_rows.map do |row|
{
warehouse_id: warehouse_id,
sku: row.fetch("sku").to_s.strip,
on_hand: row.fetch("on_hand").to_i,
reserved: row.fetch("reserved", 0).to_i,
updated_at: now
}
end
rows.reject! { |r| r[:sku].empty? }
rows.each_slice(BATCH_SIZE) do |batch|
result = InventoryLevel.upsert_counts(batch)
logger.info("reconciled #{result.rows.size} skus for warehouse=#{warehouse_id}")
end
end
end
This snippet shows how to reconcile a warehouse feed of inventory counts against the database using a single bulk upsert instead of thousands of per-row updates. The problem it solves is throughput: a periodic feed may contain tens of thousands of SKU counts, and updating each row individually issues one UPDATE per record, saturating the connection pool and holding transactions open far too long. Folding the whole batch into one statement turns that into a single round-trip that the database can plan and execute efficiently.
In InventoryLevel model, upsert_counts wraps ActiveRecord's upsert_all, which compiles to a Postgres INSERT ... ON CONFLICT DO UPDATE. The unique_by: option names the unique index that defines a conflict, and the class filters incoming rows so only whitelisted columns are written. Passing updated_at explicitly is necessary because upsert_all bypasses model callbacks and timestamp auto-management, so anything not in the payload simply will not change. The method returns the result so callers can inspect affected keys.
schema.sql defines the backing table and, crucially, the composite unique index on (warehouse_id, sku). That index is what ON CONFLICT targets — without a matching unique constraint the upsert cannot decide what counts as a duplicate and Postgres raises an error. A partial or expression index would not match unless referenced exactly.
ReconcileInventoryJob is the Sidekiq worker that assembles the payload. It normalizes each feed row into a hash, stamps a single now timestamp so the whole batch shares one write time, and slices the array into BATCH_SIZE chunks. Chunking matters because a single statement with too many rows can exceed the bind-parameter limit and balloon memory; slicing keeps each query bounded while still amortizing round-trips.
The design is naturally idempotent: replaying the same feed produces the same final counts because the conflict target collapses duplicates into updates rather than inserting new rows. A subtle trade-off is that upsert_all skips validations and callbacks, so any invariants must live in database constraints. Developers reach for this pattern whenever a job must sync large external datasets quickly and safely.
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
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
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
Share this code
Here's the card — post it anywhere.
