riverqueue
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- `by_period` uniqueness is now based off a job's `scheduled_at` instead of the current time if it has a value. [PR #39](https://github.com/riverqueue/riverqueue-ruby/pull/39).
+
 ## [0.8.0] - 2024-12-19
 
 ⚠️ Version 0.8.0 contains breaking changes to transition to River's new unique jobs implementation and to enable broader, more flexible application of unique jobs. Detailed notes on the implementation are contained in [the original River PR](https://github.com/riverqueue/river/pull/590), and the notes below include short summaries of the ways this impacts this client specifically.

diff --git a/lib/client.rb b/lib/client.rb
@@ -220,7 +220,7 @@ def insert_many(args)
       end
 
       if unique_opts.by_period
-        lower_period_bound = truncate_time(@time_now_utc.call, unique_opts.by_period).utc
+        lower_period_bound = truncate_time(insert_params.scheduled_at || @time_now_utc.call, unique_opts.by_period).utc
 
         unique_key += "&period=#{lower_period_bound.strftime("%FT%TZ")}"
       end

diff --git a/spec/client_spec.rb b/spec/client_spec.rb
@@ -291,6 +291,24 @@ def check_bigint_bounds(int)
         expect(insert_res.unique_skipped_as_duplicated).to be false
       end
 
+      it "inserts a new unique job with period determined from `scheduled_at`" do
+        job_args = ComplexArgs.new(customer_id: 1, order_id: 2, trace_id: 3, email: "john@example.com")
+        insert_opts = River::InsertOpts.new(
+          scheduled_at: now + 3600,
+          unique_opts: River::UniqueOpts.new(
+            by_period: 15 * 60
+          )
+        )
+
+        insert_res = client.insert(job_args, insert_opts: insert_opts)
+        expect(insert_res.job).to_not be_nil
+        expect(insert_res.unique_skipped_as_duplicated).to be false
+
+        unique_key_str = "&kind=#{insert_res.job.kind}" \
+          "&period=#{client.send(:truncate_time, now + 3600, 15 * 60).utc.strftime("%FT%TZ")}"
+        expect(insert_res.job.unique_key).to eq(Digest::SHA256.digest(unique_key_str))
+      end
+
       it "skips unique check if unique opts empty" do
         job_args = SimpleArgsWithInsertOpts.new(job_num: 1)
         job_args.insert_opts = River::InsertOpts.new(