hashing of tags

Author: icehaunter

packages/sync-service/lib/electric/shapes/consumer.ex

@@ -497,7 +497,12 @@ defmodule Electric.Shapes.Consumer do
 
     extra_refs = Materializer.get_all_as_refs(shape, state.stack_id)
 
-    case filter_changes(changes, shape, {xid, state.move_handling_state}, extra_refs) do
+    case filter_changes(
+           changes,
+           shape,
+           {xid, state.move_handling_state, state.stack_id, state.shape_handle},
+           extra_refs
+         ) do
       :includes_truncate ->
         # TODO: This is a very naive way to handle truncations: if ANY relevant truncates are
         #       present in the transaction, we're considering the whole transaction empty, and
@@ -656,13 +661,17 @@ defmodule Electric.Shapes.Consumer do
   defp filter_changes(
          [change | rest],
          shape,
-         {xid, filter_state} = filtering,
+         {xid, filter_state, stack_id, shape_handle} = filtering,
          extra_refs,
          change_acc,
          total_ops
        ) do
     if not MoveHandlingState.change_already_visible?(filter_state, xid, change) do
-      case Shape.convert_change(shape, change, extra_refs) do
+      case Shape.convert_change(shape, change,
+             extra_refs: extra_refs,
+             stack_id: stack_id,
+             shape_handle: shape_handle
+           ) do
         [] ->
           filter_changes(rest, shape, filtering, extra_refs, change_acc, total_ops)
 

packages/sync-service/lib/electric/shapes/consumer/snapshotter.ex

@@ -234,7 +234,7 @@ defmodule Electric.Shapes.Consumer.Snapshotter do
             }
           })
 
-        Querying.stream_initial_data(conn, stack_id, shape, chunk_bytes_threshold)
+        Querying.stream_initial_data(conn, stack_id, shape_handle, shape, chunk_bytes_threshold)
         |> Stream.transform(
           fn -> false end,
           fn item, acc ->

packages/sync-service/lib/electric/shapes/partial_modes.ex

@@ -15,7 +15,7 @@ defmodule Electric.Shapes.PartialModes do
         send(self(), {:pg_snapshot_info, pg_snapshot, lsn})
       end,
       query_fn: fn conn, _, _ ->
-        Querying.query_subset(conn, shape, subset, headers)
+        Querying.query_subset(conn, opts[:stack_id], shape_handle, shape, subset, headers)
         |> Enum.to_list()
       end,
       stack_id: opts[:stack_id],
@@ -66,7 +66,7 @@ defmodule Electric.Shapes.PartialModes do
           end,
           query_fn: fn conn, _, _ ->
             result =
-              Querying.query_move_in(conn, shape, where)
+              Querying.query_move_in(conn, opts[:stack_id], shape_handle, shape, where)
               |> results_fn.()
 
             send(parent, {:query_move_in_complete, opts[:move_in_name], result})

packages/sync-service/lib/electric/shapes/querying.ex

@@ -4,10 +4,12 @@ defmodule Electric.Shapes.Querying do
   alias Electric.Shapes.Shape
   alias Electric.Telemetry.OpenTelemetry
 
-  def query_move_in(conn, shape, {where, params}) do
+  def query_move_in(conn, stack_id, shape_handle, shape, {where, params}) do
     table = Utils.relation_to_sql(shape.root_table)
 
-    {json_like_select, _} = json_like_select(shape, %{"is_move_in" => true})
+    {json_like_select, _} =
+      json_like_select(shape, %{"is_move_in" => true}, stack_id, shape_handle)
+
     key_select = key_select(shape)
 
     query =
@@ -21,7 +23,7 @@ defmodule Electric.Shapes.Querying do
     |> Stream.flat_map(& &1.rows)
   end
 
-  def query_subset(conn, shape, subset, headers \\ []) do
+  def query_subset(conn, stack_id, shape_handle, shape, subset, headers \\ []) do
     # When querying a subset, we select same columns as the base shape
     table = Utils.relation_to_sql(shape.root_table)
 
@@ -44,7 +46,7 @@ defmodule Electric.Shapes.Querying do
     limit = if limit = subset.limit, do: " LIMIT #{limit}", else: ""
     offset = if offset = subset.offset, do: " OFFSET #{offset}", else: ""
 
-    {json_like_select, params} = json_like_select(shape, headers)
+    {json_like_select, params} = json_like_select(shape, headers, stack_id, shape_handle)
 
     query =
       Postgrex.prepare!(
@@ -80,22 +82,30 @@ defmodule Electric.Shapes.Querying do
 
   @type json_result_stream :: Enumerable.t(json_iodata())
 
-  @spec stream_initial_data(DBConnection.t(), String.t(), Shape.t(), non_neg_integer()) ::
+  @spec stream_initial_data(
+          DBConnection.t(),
+          String.t(),
+          String.t(),
+          Shape.t(),
+          non_neg_integer()
+        ) ::
           json_result_stream()
   def stream_initial_data(
         conn,
         stack_id,
+        shape_handle,
         shape,
         chunk_bytes_threshold \\ LogChunker.default_chunk_size_threshold()
       )
 
-  def stream_initial_data(_, _, %Shape{log_mode: :changes_only}, _chunk_bytes_threshold) do
+  def stream_initial_data(_, _, _, %Shape{log_mode: :changes_only}, _chunk_bytes_threshold) do
     []
   end
 
   def stream_initial_data(
         conn,
         stack_id,
+        shape_handle,
         %Shape{root_table: root_table} = shape,
         chunk_bytes_threshold
       ) do
@@ -105,7 +115,7 @@ defmodule Electric.Shapes.Querying do
       where =
         if not is_nil(shape.where), do: " WHERE " <> shape.where.query, else: ""
 
-      {json_like_select, params} = json_like_select(shape)
+      {json_like_select, params} = json_like_select(shape, [], stack_id, shape_handle)
 
       query =
         Postgrex.prepare!(conn, table, ~s|SELECT #{json_like_select} FROM #{table} #{where}|)
@@ -138,11 +148,15 @@ defmodule Electric.Shapes.Querying do
            root_table: root_table,
            selected_columns: columns
          } = shape,
-         additional_headers \\ []
+         additional_headers,
+         stack_id,
+         shape_handle
        ) do
     additional_headers =
       Shape.SubqueryMoves.move_in_tag_structure(shape)
-      |> Electric.Utils.deep_map(fn column_name -> "$${#{column_name}}" end)
+      |> Electric.Utils.deep_map(fn column_name ->
+        "$${md5('#{stack_id}#{shape_handle}' || #{column_name}::text)}"
+      end)
       |> case do
         [] -> additional_headers
         tag_structure -> additional_headers |> Map.new() |> Map.put(:tags, tag_structure)
@@ -179,9 +193,20 @@ defmodule Electric.Shapes.Querying do
     headers =
       headers
       |> Map.merge(additional_headers)
-      |> Jason.encode!()
-      |> Utils.escape_quotes(?')
-      |> String.replace(~r/\$\$\{(\w+)\}/, ~s[' || \\1::text || '])
+      |> Map.pop(:tags)
+      |> case do
+        {nil, headers} ->
+          headers |> Jason.encode!() |> Utils.escape_quotes(?')
+
+        {tags, headers} ->
+          "{" <> json = headers |> Jason.encode!() |> Utils.escape_quotes(?')
+
+          tags =
+            Jason.encode!(tags)
+            |> String.replace(~r/\$\$\{([^\}]+)\}/, ~s[' || \\1::text || '])
+
+          ~s|{"tags":#{tags},| <> json
+      end
 
     ~s['"headers":#{headers}']
   end

packages/sync-service/lib/electric/shapes/shape.ex

@@ -524,7 +524,7 @@ defmodule Electric.Shapes.Shape do
   Updates, on the other hand, may be converted to an "new record" or a "deleted record"
   if the previous/new version of the updated row isn't in the shape.
   """
-  def convert_change(shape, change, extra_refs \\ %{})
+  def convert_change(shape, change, opts \\ [])
 
   def convert_change(%__MODULE__{root_table: table}, %{relation: relation}, _)
       when table != relation,
@@ -533,13 +533,13 @@ defmodule Electric.Shapes.Shape do
   def convert_change(
         %__MODULE__{where: nil, flags: %{selects_all_columns: true}} = shape,
         change,
-        _
+        opts
       ) do
     # If the change actually doesn't change any columns, we can skip it - this is possible on Postgres but we don't care for those.
     if is_struct(change, Changes.UpdatedRecord) and change.changed_columns == MapSet.new() do
       []
     else
-      [fill_move_tags(change, shape)]
+      [fill_move_tags(change, shape, opts[:stack_id], opts[:shape_handle])]
     end
   end
 
@@ -548,14 +548,17 @@ defmodule Electric.Shapes.Shape do
   def convert_change(
         %__MODULE__{where: where, selected_columns: selected_columns} = shape,
         change,
-        extra_refs
+        opts
       )
       when is_struct(change, Changes.NewRecord)
       when is_struct(change, Changes.DeletedRecord) do
     record = if is_struct(change, Changes.NewRecord), do: change.record, else: change.old_record
 
-    if WhereClause.includes_record?(where, record, extra_refs) do
-      change |> fill_move_tags(shape) |> filter_change_columns(selected_columns) |> List.wrap()
+    if WhereClause.includes_record?(where, record, opts[:extra_refs] || %{}) do
+      change
+      |> fill_move_tags(shape, opts[:stack_id], opts[:shape_handle])
+      |> filter_change_columns(selected_columns)
+      |> List.wrap()
     else
       []
     end
@@ -564,8 +567,9 @@ defmodule Electric.Shapes.Shape do
   def convert_change(
         %__MODULE__{where: where, selected_columns: selected_columns} = shape,
         %Changes.UpdatedRecord{old_record: old_record, record: record} = change,
-        extra_refs
+        opts
       ) do
+    extra_refs = opts[:extra_refs] || %{}
     old_record_in_shape = WhereClause.includes_record?(where, old_record, extra_refs)
     new_record_in_shape = WhereClause.includes_record?(where, record, extra_refs)
 
@@ -578,7 +582,7 @@ defmodule Electric.Shapes.Shape do
       end
 
     converted_changes
-    |> Enum.map(&fill_move_tags(&1, shape))
+    |> Enum.map(&fill_move_tags(&1, shape, opts[:stack_id], opts[:shape_handle]))
     |> Enum.map(&filter_change_columns(&1, selected_columns))
     |> Enum.filter(&filtered_columns_changed?/1)
   end
@@ -589,33 +593,55 @@ defmodule Electric.Shapes.Shape do
     Changes.filter_columns(change, selected_columns)
   end
 
-  defp fill_move_tags(change, %__MODULE__{tag_structure: []}), do: change
+  defp fill_move_tags(change, %__MODULE__{tag_structure: []}, _, _), do: change
 
-  defp fill_move_tags(%Changes.NewRecord{record: record} = change, %__MODULE__{
-         tag_structure: tag_structure
-       }) do
-    move_tags = Utils.deep_map(tag_structure, fn column_name -> Map.get(record, column_name) end)
+  defp fill_move_tags(
+         %Changes.NewRecord{record: record} = change,
+         %__MODULE__{
+           tag_structure: tag_structure
+         },
+         stack_id,
+         shape_handle
+       ) do
+    move_tags = make_tags_from_pattern(tag_structure, record, stack_id, shape_handle)
     %{change | move_tags: move_tags}
   end
 
   defp fill_move_tags(
          %Changes.UpdatedRecord{record: record, old_record: old_record} = change,
-         %__MODULE__{tag_structure: tag_structure}
+         %__MODULE__{tag_structure: tag_structure},
+         stack_id,
+         shape_handle
        ) do
-    move_tags = Utils.deep_map(tag_structure, fn column_name -> Map.get(record, column_name) end)
+    move_tags = make_tags_from_pattern(tag_structure, record, stack_id, shape_handle)
 
     old_move_tags =
-      Utils.deep_map(tag_structure, fn column_name -> Map.get(old_record, column_name) end) --
+      make_tags_from_pattern(tag_structure, old_record, stack_id, shape_handle) --
         move_tags
 
     %{change | move_tags: move_tags, removed_move_tags: old_move_tags}
   end
 
-  defp fill_move_tags(%Changes.DeletedRecord{old_record: record} = change, %__MODULE__{
-         tag_structure: tag_structure
-       }) do
-    move_tags = Utils.deep_map(tag_structure, fn column_name -> Map.get(record, column_name) end)
-    %{change | move_tags: move_tags}
+  defp fill_move_tags(
+         %Changes.DeletedRecord{old_record: record} = change,
+         %__MODULE__{
+           tag_structure: tag_structure
+         },
+         stack_id,
+         shape_handle
+       ) do
+    %{change | move_tags: make_tags_from_pattern(tag_structure, record, stack_id, shape_handle)}
+  end
+
+  defp make_tags_from_pattern(pattern, record, stack_id, shape_handle) do
+    Utils.deep_map(pattern, fn column_name ->
+      make_value_hash(stack_id, shape_handle, Map.get(record, column_name))
+    end)
+  end
+
+  defp make_value_hash(stack_id, shape_handle, value) do
+    :crypto.hash(:md5, "#{stack_id}#{shape_handle}#{value}")
+    |> Base.encode16(case: :lower)
   end
 
   defp filtered_columns_changed?(%Changes.UpdatedRecord{old_record: record, record: record}),

packages/sync-service/test/electric/plug/router_test.exs

@@ -2276,23 +2276,34 @@ defmodule Electric.Plug.RouterTest do
          ]
     test "a move-in from the inner shape causes a query and new entries in the outer shape", %{
       opts: opts,
-      db_conn: db_conn
+      db_conn: db_conn,
+      stack_id: stack_id
     } do
       req = make_shape_req("child", where: "parent_id in (SELECT id FROM parent WHERE value = 1)")
-
       assert {req, 200, [data, snapshot_end]} = shape_req(req, opts)
+
+      tag =
+        :crypto.hash(:md5, stack_id <> req.handle <> "1")
+        |> Base.encode16(case: :lower)
+
       assert %{"id" => "1", "parent_id" => "1", "value" => "10"} = data["value"]
-      assert %{"operation" => "insert", "tags" => [["1"]]} = data["headers"]
+      assert %{"operation" => "insert", "tags" => [[^tag]]} = data["headers"]
       assert %{"headers" => %{"control" => "snapshot-end"}} = snapshot_end
 
       task = live_shape_req(req, opts)
 
       # Move in reflects in the new shape without invalidating it
       Postgrex.query!(db_conn, "UPDATE parent SET value = 1 WHERE id = 2", [])
 
+      tag2 =
+        :crypto.hash(:md5, stack_id <> req.handle <> "2")
+        |> Base.encode16(case: :lower)
+
       assert {_, 200, [data, %{"headers" => %{"control" => "up-to-date"}}]} = Task.await(task)
       assert %{"id" => "2", "parent_id" => "2", "value" => "20"} = data["value"]
-      assert %{"operation" => "insert", "is_move_in" => true, "tags" => [["2"]]} = data["headers"]
+
+      assert %{"operation" => "insert", "is_move_in" => true, "tags" => [[^tag2]]} =
+               data["headers"]
     end
 
     @tag with_sql: [
@@ -2439,11 +2450,15 @@ defmodule Electric.Plug.RouterTest do
       task = live_shape_req(req, ctx.opts)
       Postgrex.query!(ctx.db_conn, "INSERT INTO members (user_id, team_id) VALUES (1, 2)")
 
+      tag =
+        :crypto.hash(:md5, ctx.stack_id <> req.handle <> "2")
+        |> Base.encode16(case: :lower)
+
       assert {_, 200,
               [
                 %{
                   "value" => %{"id" => "2", "name" => "Team B"},
-                  "headers" => %{"tags" => [["2"]]}
+                  "headers" => %{"tags" => [[^tag]]}
                 },
                 _
               ]} =
@@ -2495,10 +2510,14 @@ defmodule Electric.Plug.RouterTest do
         "UPDATE members SET flag = TRUE WHERE (user_id, team_id) = (2, 2)"
       )
 
+      tag =
+        :crypto.hash(:md5, ctx.stack_id <> req.handle <> "2")
+        |> Base.encode16(case: :lower)
+
       assert {_, 200,
               [
                 %{
-                  "headers" => %{"tags" => [[["2", "2"]]]},
+                  "headers" => %{"tags" => [[[^tag, ^tag]]]},
                   "value" => %{"id" => "2", "role" => "Member"}
                 },
                 _
@@ -2541,7 +2560,11 @@ defmodule Electric.Plug.RouterTest do
       task = live_shape_req(req, ctx.opts)
       Postgrex.query!(ctx.db_conn, "UPDATE parent SET other_value = 10 WHERE id = 2")
 
-      assert {_, 200, [%{"headers" => %{"tags" => [["20"]]}, "value" => %{"id" => "3"}}, _]} =
+      tag =
+        :crypto.hash(:md5, ctx.stack_id <> req.handle <> "20")
+        |> Base.encode16(case: :lower)
+
+      assert {_, 200, [%{"headers" => %{"tags" => [[^tag]]}, "value" => %{"id" => "3"}}, _]} =
                Task.await(task)
     end
   end