chore(electric,client): Create new protocol op to represent a compensation

.changeset/itchy-carrots-invite.md

@@ -0,0 +1,6 @@
+---
+"@core/electric": patch
+"electric-sql": patch
+---
+
+[VAX-1335] Create new protocol op to represent a compensation

clients/typescript/src/_generated/protocol/satellite.ts

@@ -217,6 +217,7 @@ export interface SatTransOp {
   insert?: SatOpInsert | undefined;
   delete?: SatOpDelete | undefined;
   migrate?: SatOpMigrate | undefined;
+  compensation?: SatOpCompensation | undefined;
 }
 
 /**
@@ -299,6 +300,16 @@ export interface SatOpDelete {
   tags: string[];
 }
 
+export interface SatOpCompensation {
+  $type: "Electric.Satellite.SatOpCompensation";
+  relationId: number;
+  pkData:
+    | SatOpRow
+    | undefined;
+  /** dependency information */
+  tags: string[];
+}
+
 /** Message that corresponds to the single row. */
 export interface SatOpRow {
   $type: "Electric.Satellite.SatOpRow";
@@ -1508,6 +1519,7 @@ function createBaseSatTransOp(): SatTransOp {
     insert: undefined,
     delete: undefined,
     migrate: undefined,
+    compensation: undefined,
   };
 }
 
@@ -1533,6 +1545,9 @@ export const SatTransOp = {
     if (message.migrate !== undefined) {
       SatOpMigrate.encode(message.migrate, writer.uint32(50).fork()).ldelim();
     }
+    if (message.compensation !== undefined) {
+      SatOpCompensation.encode(message.compensation, writer.uint32(58).fork()).ldelim();
+    }
     return writer;
   },
 
@@ -1585,6 +1600,13 @@ export const SatTransOp = {
 
           message.migrate = SatOpMigrate.decode(reader, reader.uint32());
           continue;
+        case 7:
+          if (tag !== 58) {
+            break;
+          }
+
+          message.compensation = SatOpCompensation.decode(reader, reader.uint32());
+          continue;
       }
       if ((tag & 7) === 4 || tag === 0) {
         break;
@@ -1618,6 +1640,9 @@ export const SatTransOp = {
     message.migrate = (object.migrate !== undefined && object.migrate !== null)
       ? SatOpMigrate.fromPartial(object.migrate)
       : undefined;
+    message.compensation = (object.compensation !== undefined && object.compensation !== null)
+      ? SatOpCompensation.fromPartial(object.compensation)
+      : undefined;
     return message;
   },
 };
@@ -2042,6 +2067,80 @@ export const SatOpDelete = {
 
 messageTypeRegistry.set(SatOpDelete.$type, SatOpDelete);
 
+function createBaseSatOpCompensation(): SatOpCompensation {
+  return { $type: "Electric.Satellite.SatOpCompensation", relationId: 0, pkData: undefined, tags: [] };
+}
+
+export const SatOpCompensation = {
+  $type: "Electric.Satellite.SatOpCompensation" as const,
+
+  encode(message: SatOpCompensation, writer: _m0.Writer = _m0.Writer.create()): _m0.Writer {
+    if (message.relationId !== 0) {
+      writer.uint32(8).uint32(message.relationId);
+    }
+    if (message.pkData !== undefined) {
+      SatOpRow.encode(message.pkData, writer.uint32(18).fork()).ldelim();
+    }
+    for (const v of message.tags) {
+      writer.uint32(34).string(v!);
+    }
+    return writer;
+  },
+
+  decode(input: _m0.Reader | Uint8Array, length?: number): SatOpCompensation {
+    const reader = input instanceof _m0.Reader ? input : _m0.Reader.create(input);
+    let end = length === undefined ? reader.len : reader.pos + length;
+    const message = createBaseSatOpCompensation();
+    while (reader.pos < end) {
+      const tag = reader.uint32();
+      switch (tag >>> 3) {
+        case 1:
+          if (tag !== 8) {
+            break;
+          }
+
+          message.relationId = reader.uint32();
+          continue;
+        case 2:
+          if (tag !== 18) {
+            break;
+          }
+
+          message.pkData = SatOpRow.decode(reader, reader.uint32());
+          continue;
+        case 4:
+          if (tag !== 34) {
+            break;
+          }
+
+          message.tags.push(reader.string());
+          continue;
+      }
+      if ((tag & 7) === 4 || tag === 0) {
+        break;
+      }
+      reader.skipType(tag & 7);
+    }
+    return message;
+  },
+
+  create<I extends Exact<DeepPartial<SatOpCompensation>, I>>(base?: I): SatOpCompensation {
+    return SatOpCompensation.fromPartial(base ?? {});
+  },
+
+  fromPartial<I extends Exact<DeepPartial<SatOpCompensation>, I>>(object: I): SatOpCompensation {
+    const message = createBaseSatOpCompensation();
+    message.relationId = object.relationId ?? 0;
+    message.pkData = (object.pkData !== undefined && object.pkData !== null)
+      ? SatOpRow.fromPartial(object.pkData)
+      : undefined;
+    message.tags = object.tags?.map((e) => e) || [];
+    return message;
+  },
+};
+
+messageTypeRegistry.set(SatOpCompensation.$type, SatOpCompensation);
+
 function createBaseSatOpRow(): SatOpRow {
   return { $type: "Electric.Satellite.SatOpRow", nullsBitmask: new Uint8Array(), values: [] };
 }

clients/typescript/src/migrators/triggers.ts

@@ -118,10 +118,9 @@ export function generateOplogTriggers(
 /**
  * Generates triggers for compensations for all foreign keys in the provided table.
  *
- * Compensation is recorded as a specially-formatted update. It acts as a no-op, with
- * previous value set to NULL, and it's on the server to figure out that this is a no-op
- * compensation operation (usually `UPDATE` would have previous row state known). The entire
- * reason for it existing is to maybe revive the row if it has been deleted, so we need correct tags.
+ * Compensation is recorded as a SatOpCompensation messaage. The entire reason
+ * for it existing is to maybe revive the row if it has been deleted, so we need
+ * correct tags.
  *
  * The compensation update contains _just_ the primary keys, no other columns are present.
  *
@@ -155,7 +154,7 @@ function generateCompensationTriggers(table: Table): Statement[] {
              1 == (SELECT value from _electric_meta WHERE key == 'compensations')
       BEGIN
         INSERT INTO _electric_oplog (namespace, tablename, optype, primaryKey, newRow, oldRow, timestamp)
-        SELECT '${fkTableNamespace}', '${fkTableName}', 'UPDATE', json_object(${joinedFkPKs}), json_object(${joinedFkPKs}), NULL, NULL
+        SELECT '${fkTableNamespace}', '${fkTableName}', 'COMPENSATION', json_object(${joinedFkPKs}), json_object(${joinedFkPKs}), NULL, NULL
         FROM "${fkTableNamespace}"."${fkTableName}" WHERE "${foreignKey.parentKey}" = new."${foreignKey.childKey}";
       END;
       `,
@@ -167,7 +166,7 @@ function generateCompensationTriggers(table: Table): Statement[] {
               1 == (SELECT value from _electric_meta WHERE key == 'compensations')
       BEGIN
         INSERT INTO _electric_oplog (namespace, tablename, optype, primaryKey, newRow, oldRow, timestamp)
-        SELECT '${fkTableNamespace}', '${fkTableName}', 'UPDATE', json_object(${joinedFkPKs}), json_object(${joinedFkPKs}), NULL, NULL
+        SELECT '${fkTableNamespace}', '${fkTableName}', 'COMPENSATION', json_object(${joinedFkPKs}), json_object(${joinedFkPKs}), NULL, NULL
         FROM "${fkTableNamespace}"."${fkTableName}" WHERE "${foreignKey.parentKey}" = new."${foreignKey.childKey}";
       END;
       `,

clients/typescript/src/satellite/client.ts

@@ -592,6 +592,15 @@ export class SatelliteClient implements Client {
             },
           })
           break
+        case DataChangeType.COMPENSATION:
+          changeOp = SatTransOp.fromPartial({
+            compensation: {
+              pkData: record,
+              relationId: relation.id,
+              tags: tags,
+            },
+          })
+          break
       }
       ops.push(changeOp)
     })

clients/typescript/src/util/types.ts

@@ -118,6 +118,7 @@ export enum DataChangeType {
   INSERT = 'INSERT',
   UPDATE = 'UPDATE',
   DELETE = 'DELETE',
+  COMPENSATION = 'COMPENSATION',
 }
 
 export type Change = DataChange | SchemaChange

components/electric/lib/electric/postgres/shadow_table_transformation.ex

@@ -96,8 +96,7 @@ defmodule Electric.Postgres.ShadowTableTransformation do
 
   defp build_bitmask(%Changes.NewRecord{}, columns), do: Enum.map(columns, fn _ -> "t" end)
 
-  defp build_bitmask(%Changes.UpdatedRecord{old_record: nil}, columns),
-    do: Enum.map(columns, fn _ -> "f" end)
+  defp build_bitmask(%Changes.Compensation{}, columns), do: Enum.map(columns, fn _ -> "f" end)
 
   defp build_bitmask(%Changes.UpdatedRecord{old_record: old, record: new}, columns),
     do: Enum.map(columns, fn col -> if old[col] != new[col], do: "t", else: "f" end)

components/electric/lib/electric/replication/changes.ex

@@ -57,19 +57,18 @@ defmodule Electric.Replication.Changes do
     ]
 
     def count_operations(%__MODULE__{changes: changes}) do
-      base = %{operations: 0, inserts: 0, updates: 0, deletes: 0}
+      base = %{operations: 0, inserts: 0, updates: 0, deletes: 0, compensations: 0}
 
       Enum.reduce(changes, base, fn %module{}, acc ->
         key =
           case module do
             Changes.NewRecord -> :inserts
             Changes.UpdatedRecord -> :updates
             Changes.DeletedRecord -> :deletes
+            Changes.Compensation -> :compensations
           end
 
-        %{^key => value, :operations => total} = acc
-
-        %{acc | key => value + 1, :operations => total + 1}
+        Map.update!(%{acc | operations: acc.operations + 1}, key, &(&1 + 1))
       end)
     end
   end
@@ -105,6 +104,16 @@ defmodule Electric.Replication.Changes do
           }
   end
 
+  defmodule Compensation do
+    defstruct [:relation, :record, tags: []]
+
+    @type t() :: %__MODULE__{
+            relation: Changes.relation(),
+            record: Changes.record(),
+            tags: [Changes.tag()]
+          }
+  end
+
   defmodule TruncatedRelation do
     defstruct [:relation]
   end

components/electric/lib/electric/replication/postgres/slot_server.ex

@@ -458,10 +458,7 @@ defmodule Electric.Replication.Postgres.SlotServer do
     }
   end
 
-  defp changes_to_wal(
-         %Changes.UpdatedRecord{relation: table, old_record: nil, record: new},
-         relations
-       ) do
+  defp changes_to_wal(%Changes.Compensation{relation: table, record: new}, relations) do
     %ReplicationMessages.Update{
       relation_id: relations[table].oid,
       tuple_data: record_to_tuple(new, relations[table].columns)

components/electric/lib/electric/satellite/protobuf.ex

@@ -107,6 +107,7 @@ defmodule Electric.Satellite.Protobuf do
         SatOpInsert,
         SatOpUpdate,
         SatOpMigrate,
+        SatOpCompensation,
         SatTransOp,
         SatRelation,
         SatRelationColumn,
@@ -125,15 +126,6 @@ defmodule Electric.Satellite.Protobuf do
     end
   end
 
-  defmodule Version do
-    defstruct major: nil, minor: nil
-
-    @type t() :: %__MODULE__{
-            major: integer,
-            minor: integer
-          }
-  end
-
   @spec decode(byte(), binary()) :: {:ok, sq_pb_msg()} | {:error, any()}
   for {module, tag} <- @mapping do
     def decode(unquote(tag), binary) do