Module: Pgbus::Serializer

Defined in:
lib/pgbus/serializer.rb

Class Method Summary collapse

Class Method Details

.deserialize_event(json_string) ⇒ Object 



40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/pgbus/serializer.rb', line 40

def deserialize_event(json_string)
  data = JSON.parse(json_string)
  payload = data["payload"]

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

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

.deserialize_job(json_string) ⇒ Object 



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

def deserialize_job(json_string)
  Instrumentation.instrument("pgbus.serializer.deserialize", kind: :job) do
    data = JSON.parse(json_string)
    ActiveJob::Base.deserialize(data)
  end
end

.locate_global_id(gid_string) ⇒ 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.

Raises:

  • (ArgumentError) 


56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/pgbus/serializer.rb', line 56

def locate_global_id(gid_string)
  gid = GlobalID.parse(gid_string)
  raise ArgumentError, "Invalid GlobalID: #{gid_string.inspect}" unless gid

  allowed = Pgbus.configuration.allowed_global_id_models
  if allowed&.empty?
    raise ArgumentError,
          "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 ArgumentError,
          "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 ArgumentError,
          "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

  GlobalID::Locator.locate(gid)
end

.serialize_event(event) ⇒ Object 



31
32
33
34
35
36
37
38
 
# File 'lib/pgbus/serializer.rb', line 31

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 



9
10
11
12
13
14
15
16
 
# File 'lib/pgbus/serializer.rb', line 9

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 



18
19
20
21
22
 
# File 'lib/pgbus/serializer.rb', line 18

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