Fix: rename action header -> operation to match PG

docs/electric-api.yaml

@@ -202,14 +202,14 @@ paths:
 
                         Messages can be `control` messages, providing information or
                         instructions to the client. Or they can be operations that
-                        performed a certain `action` on a row of data in the shape.
+                        performed a certain `operation` on a row of data in the shape.
                       properties:
                         control:
                           type: string
                           enum:
                             - up-to-date
                             - must-refetch
-                        action:
+                        operation:
                           type: string
                           enum:
                             - insert
@@ -254,15 +254,15 @@ paths:
                         - `extra_float_digits = 1`
                 example:
                   - headers:
-                      action: insert
+                      operation: insert
                     offset: 0/0
                     key: issue-1
                     value:
                       id: issue-1
                       title: Electric
                       status: backlog
                   - headers:
-                      action: insert
+                      operation: insert
                       control: up-to-date
                     offset: 1934/0
                     key: issue-2

examples/bash-client/bash-client.bash

@@ -49,9 +49,9 @@ process_json() {
 
     # Read the JSON file line by line and save each JSON object to an individual file
     while IFS= read -r line; do
-        # Check if the headers array contains an object with key "action"
-        if echo "$line" | jq -e '.headers | map(select(.key == "action")) | length == 0' > /dev/null; then
-            # echo "Skipping line without an action: $action"  # Log skipping non-data objects
+        # Check if the headers array contains an object with key "operation"
+        if echo "$line" | jq -e '.headers | map(select(.key == "operation")) | length == 0' > /dev/null; then
+            # echo "Skipping line without an operation: $operation"  # Log skipping non-data objects
             continue
         fi
 

examples/nextjs-example/app/match-stream.ts

@@ -20,10 +20,13 @@ export async function matchStream<T>({
   return new Promise((resolve, reject) => {
     const unsubscribe = stream.subscribe((messages) => {
       for (const message of messages) {
-        if (`key` in message && operations.includes(message.headers.action)) {
+        if (
+          `key` in message &&
+          operations.includes(message.headers.operation)
+        ) {
           if (
             matchFn({
-              operationType: message.headers.action,
+              operationType: message.headers.operation,
               message: message as ChangeMessage<{ [key: string]: T }>,
             })
           ) {

examples/remix-basic/app/match-stream.ts

@@ -20,8 +20,11 @@ export async function matchStream({
   return new Promise((resolve, reject) => {
     const unsubscribe = stream.subscribe((messages) => {
       for (const message of messages) {
-        if (`key` in message && operations.includes(message.headers.action)) {
-          if (matchFn({ operationType: message.headers.action, message })) {
+        if (
+          `key` in message &&
+          operations.includes(message.headers.operation)
+        ) {
+          if (matchFn({ operationType: message.headers.operation, message })) {
             return finish(message)
           }
         }

packages/react-hooks/test/support/test-helpers.ts

@@ -36,7 +36,7 @@ export function forEachMessage<T extends Value>(
             message as Message<T>,
             messageIdx
           )
-          if (`action` in message.headers) messageIdx++
+          if (`operation` in message.headers) messageIdx++
         } catch (e) {
           controller.abort()
           return reject(e)

packages/sync-service/lib/electric/shape_cache/cub_db_storage.ex

@@ -204,12 +204,12 @@ defmodule Electric.ShapeCache.CubDbStorage do
     {snapshot_key(shape_id, index), {change_key, :insert, serialized_row, %{}}}
   end
 
-  defp storage_item_to_log_item({key, {change_key, action, value, header_data}})
+  defp storage_item_to_log_item({key, {change_key, operation, value, header_data}})
        when is_binary(change_key) do
     %{
       key: change_key,
       value: value,
-      headers: Map.put(header_data, :action, action),
+      headers: Map.put(header_data, :operation, operation),
       offset: offset(key)
     }
   end

packages/sync-service/lib/electric/shape_cache/in_memory_storage.ex

@@ -106,9 +106,9 @@ defmodule Electric.ShapeCache.InMemoryStorage do
     ets_table = opts.log_ets_table
 
     changes
-    |> Enum.map(fn {offset, key, action, value, header_data} ->
+    |> Enum.map(fn {offset, key, operation, value, header_data} ->
       offset = storage_offset(offset)
-      {{shape_id, offset}, key, action, value, header_data}
+      {{shape_id, offset}, key, operation, value, header_data}
     end)
     |> then(&:ets.insert(ets_table, &1))
 
@@ -141,15 +141,24 @@ defmodule Electric.ShapeCache.InMemoryStorage do
   end
 
   defp snapshot_storage_item_to_log_item({_shape_id, snapshot_log_offset}, {key, value}) do
-    %{key: key, value: value, headers: %{action: :insert}, offset: snapshot_log_offset}
+    %{key: key, value: value, headers: %{operation: :insert}, offset: snapshot_log_offset}
   end
 
   # Turns a stored log item into a log item
   # by modifying the turning the tuple offset
   # back into a LogOffset value.
-  defp log_storage_item_to_log_item({_shape_id, position}, {_, key, action, value, header_data}) do
+  defp log_storage_item_to_log_item(
+         {_shape_id, position},
+         {_, key, operation, value, header_data}
+       ) do
     offset = LogOffset.new(position)
-    %{key: key, value: value, headers: Map.put(header_data, :action, action), offset: offset}
+
+    %{
+      key: key,
+      value: value,
+      headers: Map.put(header_data, :operation, operation),
+      offset: offset
+    }
   end
 
   # Turns a LogOffset into a tuple representation

packages/sync-service/lib/electric/shape_cache/storage.ex

@@ -13,7 +13,7 @@ defmodule Electric.ShapeCache.Storage do
           {
             offset :: LogOffset.t(),
             key :: String.t(),
-            action :: :insert | :update | :delete,
+            operation :: :insert | :update | :delete,
             value :: %{String.t() => String.t()},
             header_data :: %{optional(atom()) => Jason.Encoder.t()}
           }

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

@@ -55,7 +55,7 @@ defmodule Electric.Plug.RouterTest do
 
       assert [
                %{
-                 "headers" => %{"action" => "insert"},
+                 "headers" => %{"operation" => "insert"},
                  "key" => _,
                  "offset" => @first_offset,
                  "value" => %{
@@ -159,7 +159,7 @@ defmodule Electric.Plug.RouterTest do
 
       assert [
                %{
-                 "headers" => %{"action" => "insert"},
+                 "headers" => %{"operation" => "insert"},
                  "key" => ^key,
                  "offset" => @first_offset,
                  "value" => %{
@@ -198,7 +198,7 @@ defmodule Electric.Plug.RouterTest do
 
       assert [
                %{
-                 "headers" => %{"action" => "insert"},
+                 "headers" => %{"operation" => "insert"},
                  "key" => ^key2,
                  "offset" => _,
                  "value" => %{
@@ -290,17 +290,17 @@ defmodule Electric.Plug.RouterTest do
 
       assert [
                %{
-                 "headers" => %{"action" => "delete"},
+                 "headers" => %{"operation" => "delete"},
                  "value" => %{"id" => "1"},
                  "key" => ^key
                },
                %{
-                 "headers" => %{"action" => "insert"},
+                 "headers" => %{"operation" => "insert"},
                  "value" => %{"id" => "2", "value1" => _, "value2" => _, "value3" => _},
                  "key" => key2
                },
                %{
-                 "headers" => %{"action" => "insert"},
+                 "headers" => %{"operation" => "insert"},
                  "value" => %{"id" => "3", "value1" => _, "value2" => _, "value3" => _},
                  "key" => key3
                },
@@ -341,17 +341,17 @@ defmodule Electric.Plug.RouterTest do
 
       assert [
                %{
-                 "headers" => %{"action" => "delete"},
+                 "headers" => %{"operation" => "delete"},
                  "value" => %{"col1" => "test1", "col2" => "test2"},
                  "key" => ^key
                },
                %{
-                 "headers" => %{"action" => "insert"},
+                 "headers" => %{"operation" => "insert"},
                  "value" => %{"col1" => "test3", "col2" => "test2"},
                  "key" => key2
                },
                %{
-                 "headers" => %{"action" => "insert"},
+                 "headers" => %{"operation" => "insert"},
                  "value" => %{"col1" => "test4", "col2" => "test5"},
                  "key" => key3
                },

packages/sync-service/test/electric/shape_cache/storage_implementations_test.exs

@@ -91,13 +91,13 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
                    offset: @snapshot_offset,
                    value: %{"id" => "00000000-0000-0000-0000-000000000001", "title" => "row1"},
                    key: ~S|"public"."the-table"/"00000000-0000-0000-0000-000000000001"|,
-                   headers: %{action: :insert}
+                   headers: %{operation: :insert}
                  },
                  %{
                    offset: @snapshot_offset,
                    value: %{"id" => "00000000-0000-0000-0000-000000000002", "title" => "row2"},
                    key: ~S|"public"."the-table"/"00000000-0000-0000-0000-000000000002"|,
-                   headers: %{action: :insert}
+                   headers: %{operation: :insert}
                  }
                ] = Enum.to_list(stream)
       end
@@ -125,13 +125,13 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
                    offset: @snapshot_offset,
                    value: %{"id" => "00000000-0000-0000-0000-000000000001", "title" => "row1"},
                    key: ~S|"public"."the-table"/"00000000-0000-0000-0000-000000000001"|,
-                   headers: %{action: :insert}
+                   headers: %{operation: :insert}
                  },
                  %{
                    offset: @snapshot_offset,
                    value: %{"id" => "00000000-0000-0000-0000-000000000002", "title" => "row2"},
                    key: ~S|"public"."the-table"/"00000000-0000-0000-0000-000000000002"|,
-                   headers: %{action: :insert}
+                   headers: %{operation: :insert}
                  }
                ] = Enum.to_list(stream)
       end
@@ -188,7 +188,7 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
 
         assert entry.key == ~S|"public"."test_table"/"123"|
         assert entry.value == %{"id" => "123", "name" => "Test"}
-        assert entry.headers == %{action: :insert, txid: 1, relation: ["public", "test_table"]}
+        assert entry.headers == %{operation: :insert, txid: 1, relation: ["public", "test_table"]}
         assert entry.offset == offset
       end
     end
@@ -238,9 +238,9 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
         entries = Enum.to_list(stream)
 
         assert [
-                 %{headers: %{action: :insert}},
-                 %{headers: %{action: :update}},
-                 %{headers: %{action: :delete}}
+                 %{headers: %{operation: :insert}},
+                 %{headers: %{operation: :update}},
+                 %{headers: %{operation: :delete}}
                ] = entries
       end
 
@@ -283,8 +283,8 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
         entries = Enum.to_list(stream)
 
         assert [
-                 %{headers: %{action: :update}},
-                 %{headers: %{action: :delete}}
+                 %{headers: %{operation: :update}},
+                 %{headers: %{operation: :delete}}
                ] = entries
       end
 
@@ -334,7 +334,7 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
 
         entries = Enum.to_list(stream)
 
-        assert [%{headers: %{action: :update}}] = entries
+        assert [%{headers: %{operation: :update}}] = entries
       end
 
       test "returns only logs for the requested shape_id", %{module: storage, opts: opts} do

packages/typescript-client/src/client.ts

@@ -516,10 +516,10 @@ export class Shape {
     messages.forEach((message) => {
       if (`key` in message) {
         dataMayHaveChanged = [`insert`, `update`, `delete`].includes(
-          message.headers.action
+          message.headers.operation
         )
 
-        switch (message.headers.action) {
+        switch (message.headers.operation) {
           case `insert`:
             this.data.set(message.key, message.value)
             break

packages/typescript-client/src/types.ts

@@ -20,7 +20,7 @@ export type ControlMessage = {
 export type ChangeMessage<T> = {
   key: string
   value: T
-  headers: Header & { action: `insert` | `update` | `delete` }
+  headers: Header & { operation: `insert` | `update` | `delete` }
   offset: Offset
 }
 

packages/typescript-client/test/support/test-helpers.ts

@@ -37,7 +37,7 @@ export function forEachMessage<T extends Value>(
             message as Message<T>,
             messageIdx
           )
-          if (`action` in message.headers) messageIdx++
+          if (`operation` in message.headers) messageIdx++
         } catch (e) {
           controller.abort()
           return reject(e)