Module: Pgbus::Serializer

Defined in:
lib/pgbus/serializer.rb

Class Method Summary collapse

Class Method Details

.assert_allowed_global_id!(gid_string, configuration: Pgbus.configuration) ⇒ Object

Raises SerializationError unless the GlobalID's model is permitted. Returns the parsed GlobalID on success (so locate can skip re-parse).

Raises:



83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/pgbus/serializer.rb', line 83

def assert_allowed_global_id!(gid_string, configuration: Pgbus.configuration)
  gid = GlobalID.parse(gid_string)
  raise Pgbus::SerializationError, "Invalid GlobalID: #{gid_string.inspect}" unless gid

  allowed = configuration.allowed_global_id_models
  if allowed && allowed.empty?
    raise Pgbus::SerializationError,
          "GlobalID deserialization is disabled (allowed_global_id_models is empty). " \
          "Set to nil to allow all models, or add permitted classes."
  end
  if allowed&.any? { |entry| !entry.is_a?(Class) && !entry.is_a?(Module) }
    raise Pgbus::SerializationError,
          "allowed_global_id_models must contain Class/Module objects, " \
          "got: #{allowed.map(&:class).uniq.join(", ")}"
  end
  if allowed&.none? { |klass| gid.model_class <= klass }
    raise Pgbus::SerializationError,
          "GlobalID model #{gid.model_class} is not in allowed_global_id_models. " \
          "Add it to Pgbus.configuration.allowed_global_id_models to permit deserialization."
  end

  gid
end

.assert_job_global_ids_allowed!(data, configuration: Pgbus.configuration) ⇒ Object

Walk a job payload (or any nested structure) and enforce the allowlist on every ActiveJob _aj_globalid value. No-op when allowlist is nil.



109
110
111
112
113
 
# File 'lib/pgbus/serializer.rb', line 109

def assert_job_global_ids_allowed!(data, configuration: Pgbus.configuration)
  return if configuration.allowed_global_id_models.nil?

  walk_job_global_ids(data, configuration)
end

.deserialize_event(json_string, configuration: Pgbus.configuration) ⇒ Object 



57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/pgbus/serializer.rb', line 57

def deserialize_event(json_string, configuration: Pgbus.configuration)
  data = JSON.parse(json_string)
  payload = data["payload"]

  if payload.is_a?(Hash) && payload["_global_id"]
    data["payload"] = locate_global_id(payload["_global_id"], configuration: configuration)
  end

  Event.new(
    event_id: data["event_id"],
    payload: data["payload"],
    published_at: Time.parse(data["published_at"])
  )
end

.deserialize_job(json_string, configuration: Pgbus.configuration) ⇒ Object 



29
30
31
32
33
 
# File 'lib/pgbus/serializer.rb', line 29

def deserialize_job(json_string, configuration: Pgbus.configuration)
  Instrumentation.instrument("pgbus.serializer.deserialize", kind: :job) do
    deserialize_job_data(JSON.parse(json_string), configuration: configuration)
  end
end

.deserialize_job_data(data, configuration: Pgbus.configuration) ⇒ Object

Job-hash entry point used by the executor (payload already parsed) and by deserialize_job. When allowed_global_id_models is configured, every _aj_globalid in the tree is checked before Rails' unrestricted GlobalID::Locator runs (issue #368). Nil allowlist = zero-cost allow-all. Prefer the caller's configuration (e.g. Executor's injected config) so a non-global allowlist is not silently ignored.



41
42
43
44
45
46
 
# File 'lib/pgbus/serializer.rb', line 41

def deserialize_job_data(data, configuration: Pgbus.configuration)
  assert_job_global_ids_allowed!(data, configuration: configuration)
  # Top-level constant: bare ActiveJob::Base can resolve to
  # Pgbus::ActiveJob::Base under Zeitwerk's Pgbus::ActiveJob namespace.
  ::ActiveJob::Base.deserialize(data)
end

.locate_global_id(gid_string, configuration: Pgbus.configuration) ⇒ Object

Locate a GlobalID with optional type restriction. When allowed_global_id_models is configured, only those model classes can be resolved — prevents loading arbitrary objects from crafted payloads. Shared by EventBus payloads (_global_id) and job arguments (_aj_globalid).



76
77
78
79
 
# File 'lib/pgbus/serializer.rb', line 76

def locate_global_id(gid_string, configuration: Pgbus.configuration)
  gid = assert_allowed_global_id!(gid_string, configuration: configuration)
  GlobalID::Locator.locate(gid)
end

.serialize_event(event) ⇒ Object 



48
49
50
51
52
53
54
55
 
# File 'lib/pgbus/serializer.rb', line 48

def serialize_event(event)
  payload = event.respond_to?(:to_global_id) ? { "_global_id" => event.to_global_id.to_s } : event
  JSON.generate({
                  "event_id" => event.respond_to?(:event_id) ? event.event_id : SecureRandom.uuid,
                  "payload" => payload,
                  "published_at" => Time.now.utc.iso8601(6)
                })
end

.serialize_job(active_job) ⇒ Object 



14
15
16
17
18
19
20
21
 
# File 'lib/pgbus/serializer.rb', line 14

def serialize_job(active_job)
  Instrumentation.instrument("pgbus.serializer.serialize", kind: :job) do
    data = active_job.serialize
    # GlobalID is handled by ActiveJob's serialize — it converts AR objects
    # to GlobalID URIs automatically. We just JSON-encode the result.
    JSON.generate(data)
  end
end

.serialize_job_hash(active_job) ⇒ Object 



23
24
25
26
27
 
# File 'lib/pgbus/serializer.rb', line 23

def serialize_job_hash(active_job)
  Instrumentation.instrument("pgbus.serializer.serialize", kind: :job) do
    active_job.serialize
  end
end