Fix ActiveRecord JSONB args (#60)

* Fix ActiveRecord inserting job args as JSONB strings instead of objects

The ActiveRecord driver passed the JSON-encoded args string directly to
the insert. ActiveRecord's JSONB type serializer JSON-encodes all values
on write, so a string like '{"job_num":1}' gets double-encoded into a
JSONB string literal rather than stored as a JSONB object.

Go workers then fail to unmarshal the double-encoded value:

    json: cannot unmarshal string into Go value of type ...Args

Parse the encoded args to a Hash before insert so ActiveRecord
serializes it as a proper JSONB object. Add a deserialize_json helper
on read paths for backwards compatibility with existing fixture data.

* Expect JSONB object values on Ruby read paths, same as Go

Remove the deserialize_json helper that tolerated double-encoded string
values. ActiveRecord natively deserializes JSONB objects to Hashes, so
no parsing is needed. Update test fixtures to use Hash args matching
what ActiveRecord returns from properly stored JSONB objects.

Author: primdahl

driver/riverqueue-activerecord/lib/driver.rb

@@ -71,7 +71,7 @@ def transaction(&)
 
     private def insert_params_to_hash(insert_params)
       {
-        args: insert_params.encoded_args,
+        args: JSON.parse(insert_params.encoded_args),
         kind: insert_params.kind,
         max_attempts: insert_params.max_attempts,
         priority: insert_params.priority,
@@ -91,7 +91,7 @@ def transaction(&)
 
       River::JobRow.new(
         id: river_job.id,
-        args: JSON.parse(river_job.args),
+        args: river_job.args,
         attempt: river_job.attempt,
         attempted_at: river_job.attempted_at&.getutc,
         attempted_by: river_job.attempted_by,
@@ -153,7 +153,7 @@ def transaction(&)
       [
         River::JobRow.new(
           id: river_job["id"],
-          args: JSON.parse(river_job["args"]),
+          args: river_job["args"],
           attempt: river_job["attempt"],
           attempted_at: river_job["attempted_at"]&.getutc,
           attempted_by: river_job["attempted_by"],

driver/riverqueue-activerecord/spec/driver_spec.rb

@@ -15,11 +15,26 @@
 
   it_behaves_like "driver shared examples"
 
+  describe "client inserts" do
+    it "persists args as a JSON object rather than a JSON string" do
+      insert_res = client.insert(SimpleArgs.new(job_num: 1))
+
+      row = ActiveRecord::Base.connection.exec_query(<<~SQL).first
+        SELECT args, jsonb_typeof(args) AS args_type
+        FROM river_job
+        WHERE id = #{insert_res.job.id}
+      SQL
+
+      expect(row["args_type"]).to eq("object")
+      expect(JSON.parse(row["args"])).to eq({"job_num" => 1})
+    end
+  end
+
   describe "#to_job_row_from_model" do
     it "converts a database record to `River::JobRow` with minimal properties" do
       river_job = River::Driver::ActiveRecord::RiverJob.create(
         id: 1,
-        args: %({"job_num":1}),
+        args: {"job_num" => 1},
         kind: "simple",
         max_attempts: River::MAX_ATTEMPTS_DEFAULT,
         priority: River::PRIORITY_DEFAULT,
@@ -56,7 +71,7 @@
         attempted_at: now,
         attempted_by: ["client1"],
         created_at: now,
-        args: %({"job_num":1}),
+        args: {"job_num" => 1},
         finalized_at: now,
         kind: "simple",
         max_attempts: River::MAX_ATTEMPTS_DEFAULT,
@@ -93,7 +108,7 @@
     it "with errors" do
       now = Time.now.utc
       river_job = River::Driver::ActiveRecord::RiverJob.create(
-        args: %({"job_num":1}),
+        args: {"job_num" => 1},
         errors: [JSON.dump(
           {
             at: now,
@@ -124,7 +139,7 @@
     it "converts a database record to `River::JobRow` with minimal properties" do
       res = River::Driver::ActiveRecord::RiverJob.insert({
         id: 1,
-        args: %({"job_num":1}),
+        args: {"job_num" => 1},
         kind: "simple",
         max_attempts: River::MAX_ATTEMPTS_DEFAULT
       }, returning: Arel.sql("*, false AS unique_skipped_as_duplicate"))
@@ -159,7 +174,7 @@
         attempted_at: now,
         attempted_by: ["client1"],
         created_at: now,
-        args: %({"job_num":1}),
+        args: {"job_num" => 1},
         finalized_at: now,
         kind: "simple",
         max_attempts: River::MAX_ATTEMPTS_DEFAULT,
@@ -197,7 +212,7 @@
     it "with errors" do
       now = Time.now.utc
       res = River::Driver::ActiveRecord::RiverJob.insert({
-        args: %({"job_num":1}),
+        args: {"job_num" => 1},
         errors: [JSON.dump(
           {
             at: now,

driver/riverqueue-sequel/spec/driver_spec.rb

@@ -9,6 +9,21 @@
 
   it_behaves_like "driver shared examples"
 
+  describe "client inserts" do
+    it "persists args as a JSON object rather than a JSON string" do
+      insert_res = client.insert(SimpleArgs.new(job_num: 1))
+
+      row = DB.fetch(<<~SQL, insert_res.job.id).first
+        SELECT args, jsonb_typeof(args) AS args_type
+        FROM river_job
+        WHERE id = ?
+      SQL
+
+      expect(row[:args_type]).to eq("object")
+      expect(row[:args].to_h).to eq({"job_num" => 1})
+    end
+  end
+
   describe "#to_job_row" do
     it "converts a database record to `River::JobRow` with minimal properties" do
       river_job = DB[:river_job].returning.insert_select({