Metadata serializer on top of 1.4.4

config/bench.exs

@@ -13,7 +13,8 @@ default_config = [
   database: "eventstore_bench",
   hostname: "localhost",
   pool_size: 10,
-  serializer: EventStore.TermSerializer
+  serializer: EventStore.TermSerializer,
+  metadata_serializer: EventStore.TermSerializer
 ]
 
 config :eventstore, TestEventStore, default_config

config/jsonb.exs

@@ -9,12 +9,14 @@ config :ex_unit,
 
 default_config = [
   column_data_type: "jsonb",
+  metadata_column_data_type: "jsonb",
   username: "postgres",
   password: "postgres",
   database: "eventstore_jsonb_test",
   hostname: "localhost",
   pool_size: 1,
   serializer: EventStore.JsonbSerializer,
+  metadata_serializer: EventStore.JsonbSerializer,
   subscription_retry_interval: 1_000,
   types: EventStore.PostgresTypes
 ]

config/migration.exs

@@ -14,6 +14,7 @@ default_config = [
   hostname: "localhost",
   pool_size: 1,
   serializer: EventStore.JsonSerializer,
+  metadata_serializer: EventStore.JsonSerializer,
   subscription_retry_interval: 1_000
 ]
 

config/test.exs

@@ -15,6 +15,7 @@ default_config = [
   hostname: "localhost",
   pool_size: 1,
   serializer: EventStore.JsonSerializer,
+  metadata_serializer: EventStore.JsonSerializer,
   subscription_retry_interval: 1_000
 ]
 

lib/event_store.ex

@@ -23,6 +23,7 @@ defmodule EventStore do
 
       config :my_app, MyApp.EventStore,
         serializer: EventStore.JsonSerializer,
+        metadata_serializer: EventStore.JsonSerializer,
         username: "postgres",
         password: "postgres",
         database: "eventstore",
@@ -448,6 +449,7 @@ defmodule EventStore do
         conn = Keyword.fetch!(config, :conn)
         schema = Keyword.fetch!(config, :schema)
         serializer = Keyword.fetch!(config, :serializer)
+        metadata_serializer = Keyword.fetch!(config, :metadata_serializer)
 
         query_timeout = timeout(opts, config)
 
@@ -467,6 +469,7 @@ defmodule EventStore do
               query_timeout: query_timeout,
               schema: schema,
               serializer: serializer,
+              metadata_serializer: metadata_serializer,
               stream_uuid: stream_uuid,
               subscription_name: subscription_name,
               start_from: start_from

lib/event_store/config.ex

@@ -67,6 +67,19 @@ defmodule EventStore.Config do
     end
   end
 
+  def metadata_column_data_type(event_store, config) do
+    case Keyword.get(config, :metadata_column_data_type, column_data_type(event_store, config)) do
+      valid when valid in ["bytea", "jsonb"] ->
+        valid
+
+      invalid ->
+        raise ArgumentError,
+              inspect(event_store) <>
+                " `:column_data_type` expects either \"bytea\" or \"jsonb\" but got: " <>
+                inspect(invalid)
+    end
+  end
+
   @postgrex_connection_opts [
     :after_connect,
     :after_connect_timeout,

lib/event_store/config/parser.ex

@@ -3,6 +3,7 @@ defmodule EventStore.Config.Parser do
 
   @config_defaults [
     column_data_type: "bytea",
+    metadata_column_data_type: "bytea",
     enable_hard_deletes: false,
     schema: "public",
     timeout: 15_000

lib/event_store/notifications/publisher.ex

@@ -12,7 +12,15 @@ defmodule EventStore.Notifications.Publisher do
   alias EventStore.Notifications.Notification
 
   defmodule State do
-    defstruct [:conn, :event_store, :query_timeout, :schema, :serializer, :subscribe_to]
+    defstruct [
+      :conn,
+      :event_store,
+      :query_timeout,
+      :schema,
+      :serializer,
+      :metadata_serializer,
+      :subscribe_to
+    ]
 
     def new(opts) do
       %State{
@@ -21,6 +29,7 @@ defmodule EventStore.Notifications.Publisher do
         query_timeout: Keyword.fetch!(opts, :query_timeout),
         schema: Keyword.fetch!(opts, :schema),
         serializer: Keyword.fetch!(opts, :serializer),
+        metadata_serializer: Keyword.fetch!(opts, :metadata_serializer),
         subscribe_to: Keyword.fetch!(opts, :subscribe_to)
       }
     end
@@ -63,7 +72,13 @@ defmodule EventStore.Notifications.Publisher do
       to_stream_version: to_stream_version
     } = notification
 
-    %State{conn: conn, query_timeout: query_timeout, schema: schema, serializer: serializer} =
+    %State{
+      conn: conn,
+      query_timeout: query_timeout,
+      schema: schema,
+      serializer: serializer,
+      metadata_serializer: metadata_serializer
+    } =
       state
 
     count = to_stream_version - from_stream_version + 1
@@ -74,7 +89,8 @@ defmodule EventStore.Notifications.Publisher do
              timeout: query_timeout
            ) do
         {:ok, events} ->
-          deserialized_events = deserialize_recorded_events(events, serializer)
+          deserialized_events =
+            deserialize_recorded_events(events, serializer, metadata_serializer)
 
           {stream_uuid, deserialized_events}
 
@@ -92,8 +108,8 @@ defmodule EventStore.Notifications.Publisher do
     end
   end
 
-  defp deserialize_recorded_events(recorded_events, serializer) do
-    Enum.map(recorded_events, &RecordedEvent.deserialize(&1, serializer))
+  defp deserialize_recorded_events(recorded_events, serializer, metadata_serializer) do
+    Enum.map(recorded_events, &RecordedEvent.deserialize(&1, serializer, metadata_serializer))
   end
 
   defp broadcast(event_store, stream_uuid, events) do

lib/event_store/notifications/supervisor.ex

@@ -22,6 +22,7 @@ defmodule EventStore.Notifications.Supervisor do
     conn = Keyword.fetch!(config, :conn)
     schema = Keyword.fetch!(config, :schema)
     serializer = Keyword.fetch!(config, :serializer)
+    metadata_serializer = Keyword.fetch!(config, :metadata_serializer)
     query_timeout = Keyword.fetch!(config, :timeout)
 
     listener_name = Module.concat([event_store, Listener])
@@ -54,6 +55,7 @@ defmodule EventStore.Notifications.Supervisor do
          event_store: event_store,
          schema: schema,
          serializer: serializer,
+         metadata_serializer: metadata_serializer,
          subscribe_to: listener_name,
          name: publisher_name,
          hibernate_after: hibernate_after}

lib/event_store/recorded_event.ex

@@ -55,13 +55,13 @@ defmodule EventStore.RecordedEvent do
     :created_at
   ]
 
-  def deserialize(%RecordedEvent{} = recorded_event, serializer) do
+  def deserialize(%RecordedEvent{} = recorded_event, serializer, metadata_serializer) do
     %RecordedEvent{data: data, metadata: metadata, event_type: event_type} = recorded_event
 
     %RecordedEvent{
       recorded_event
       | data: serializer.deserialize(data, type: event_type),
-        metadata: serializer.deserialize(metadata, [])
+        metadata: metadata_serializer.deserialize(metadata, [])
     }
   end
 

lib/event_store/serializer.ex

@@ -31,4 +31,19 @@ defmodule EventStore.Serializer do
           message: "#{inspect(event_store)} configuration expects :serializer to be configured"
     end
   end
+
+  @doc """
+  Get the metadata serializer module from the given config for the event store.
+  """
+  def metadata_serializer(event_store, config) do
+    case Keyword.fetch(config, :metadata_serializer) do
+      {:ok, serializer} ->
+        serializer
+
+      :error ->
+        raise ArgumentError,
+          message:
+            "#{inspect(event_store)} configuration expects :metadata_serializer to be configured"
+    end
+  end
 end

lib/event_store/snapshots/snapshot_data.ex

@@ -16,23 +16,23 @@ defmodule EventStore.Snapshots.SnapshotData do
           created_at: DateTime.t()
         }
 
-  def serialize(%SnapshotData{} = snapshot, serializer) do
+  def serialize(%SnapshotData{} = snapshot, serializer, metadata_serializer) do
     %SnapshotData{data: data, metadata: metadata} = snapshot
 
     %SnapshotData{
       snapshot
       | data: serializer.serialize(data),
-        metadata: serializer.serialize(metadata)
+        metadata: metadata_serializer.serialize(metadata)
     }
   end
 
-  def deserialize(%SnapshotData{} = snapshot, serializer) do
+  def deserialize(%SnapshotData{} = snapshot, serializer, metadata_serializer) do
     %SnapshotData{source_type: source_type, data: data, metadata: metadata} = snapshot
 
     %SnapshotData{
       snapshot
       | data: serializer.deserialize(data, type: source_type),
-        metadata: serializer.deserialize(metadata, [])
+        metadata: metadata_serializer.deserialize(metadata, [])
     }
   end
 end

lib/event_store/snapshots/snapshotter.ex

@@ -9,9 +9,10 @@ defmodule EventStore.Snapshots.Snapshotter do
   """
   def read_snapshot(conn, source_uuid, opts) do
     {serializer, opts} = Keyword.pop(opts, :serializer)
+    {metadata_serializer, opts} = Keyword.pop(opts, :metadata_serializer)
 
     with {:ok, snapshot} <- Snapshot.read_snapshot(conn, source_uuid, opts) do
-      deserialized = SnapshotData.deserialize(snapshot, serializer)
+      deserialized = SnapshotData.deserialize(snapshot, serializer, metadata_serializer)
 
       {:ok, deserialized}
     end
@@ -24,8 +25,9 @@ defmodule EventStore.Snapshots.Snapshotter do
   """
   def record_snapshot(conn, %SnapshotData{} = snapshot, opts) do
     {serializer, opts} = Keyword.pop(opts, :serializer)
+    {metadata_serializer, opts} = Keyword.pop(opts, :metadata_serializer)
 
-    serialized = SnapshotData.serialize(snapshot, serializer)
+    serialized = SnapshotData.serialize(snapshot, serializer, metadata_serializer)
 
     Snapshot.record_snapshot(conn, serialized, opts)
   end

lib/event_store/sql/init.ex

@@ -5,13 +5,14 @@ defmodule EventStore.Sql.Init do
 
   def statements(config) do
     column_data_type = Keyword.fetch!(config, :column_data_type)
+    metadata_column_data_type = Keyword.fetch!(config, :metadata_column_data_type)
     schema = Keyword.fetch!(config, :schema)
 
     [
       ~s(SET LOCAL search_path TO "#{schema}";),
       create_streams_table(),
       create_stream_uuid_index(),
-      create_events_table(column_data_type),
+      create_events_table(column_data_type, metadata_column_data_type),
       create_stream_events_table(),
       create_stream_events_index(),
       create_event_store_exception_function(),
@@ -26,7 +27,7 @@ defmodule EventStore.Sql.Init do
       create_event_notification_trigger(),
       create_subscriptions_table(),
       create_subscription_index(),
-      create_snapshots_table(column_data_type),
+      create_snapshots_table(column_data_type, metadata_column_data_type),
       create_schema_migrations_table(),
       record_event_store_schema_version()
     ]
@@ -58,7 +59,7 @@ defmodule EventStore.Sql.Init do
     """
   end
 
-  defp create_events_table(column_data_type) do
+  defp create_events_table(column_data_type, metadata_column_data_type) do
     """
     CREATE TABLE events
     (
@@ -67,7 +68,7 @@ defmodule EventStore.Sql.Init do
         causation_id uuid NULL,
         correlation_id uuid NULL,
         data #{column_data_type} NOT NULL,
-        metadata #{column_data_type} NULL,
+        metadata #{metadata_column_data_type} NULL,
         created_at timestamp with time zone DEFAULT NOW() NOT NULL
     );
     """
@@ -242,15 +243,15 @@ defmodule EventStore.Sql.Init do
     """
   end
 
-  defp create_snapshots_table(column_data_type) do
+  defp create_snapshots_table(column_data_type, metadata_column_data_type) do
     """
     CREATE TABLE snapshots
     (
         source_uuid text PRIMARY KEY NOT NULL,
         source_version bigint NOT NULL,
         source_type text NOT NULL,
         data #{column_data_type} NOT NULL,
-        metadata #{column_data_type} NULL,
+        metadata #{metadata_column_data_type} NULL,
         created_at timestamp with time zone DEFAULT NOW() NOT NULL
     );
     """