Serialise grants and assigns to protobuf

closes:

- [VAX-1644](https://linear.app/electric-sql/issue/VAX-1644/serialise-grant-statements-to-protobuf)
- [VAX-1647](https://linear.app/electric-sql/issue/VAX-1647/serialise-assign-statements-to-pb)

components/electric/config/runtime.exs

@@ -182,7 +182,8 @@ config :electric, Electric.Features,
   proxy_ddlx_grant: false,
   proxy_ddlx_revoke: false,
   proxy_ddlx_assign: false,
-  proxy_ddlx_unassign: false
+  proxy_ddlx_unassign: false,
+  proxy_ddlx_sqlite: false
 
 {:ok, conn_params} = database_url_config
 

components/electric/lib/electric/ddlx/command.ex

@@ -1,31 +1,238 @@
-defprotocol Electric.DDLX.Command do
-  @spec pg_sql(t()) :: [String.t()]
-  def pg_sql(command)
+defprotocol Electric.DDLX.Command.PgSQL do
+  @spec to_sql(t()) :: [String.t()]
+  def to_sql(cmd)
+end
+
+alias Electric.Satellite.SatPerms
+
+defmodule Electric.DDLX.Command do
+  alias Electric.DDLX
+  alias Electric.DDLX.Command.PgSQL
+
+  defstruct [:cmds, :stmt, :tag, tables: []]
+
+  @type t() :: %__MODULE__{
+          cmds: struct(),
+          stmt: String.t(),
+          tag: String.t(),
+          tables: [Electric.Postgres.relation()]
+        }
+
+  def tag(%__MODULE__{tag: tag}), do: tag
+
+  def pg_sql(cmd) do
+    PgSQL.to_sql(cmd)
+  end
+
+  def table_names(%__MODULE__{tables: tables}), do: tables
+
+  def enabled?(%__MODULE__{cmds: cmd}) do
+    command_enabled?(cmd)
+  end
+
+  def electric_enable({_, _} = table) do
+    table_name = Electric.Utils.inspect_relation(table)
+
+    %__MODULE__{
+      tag: "ELECTRIC ENABLE",
+      tables: [table],
+      cmds: %DDLX.Command.Enable{table_name: table_name},
+      stmt: "CALL electric.electrify('#{table_name}');"
+    }
+  end
+
+  # shortcut the enable command, which has to be enabled
+  defp command_enabled?(%DDLX.Command.Enable{}), do: true
+  defp command_enabled?(%DDLX.Command.Disable{}), do: false
+
+  defp command_enabled?(%SatPerms.DDLX{} = ddlx) do
+    ddlx
+    |> command_list()
+    |> Enum.map(&feature_flag/1)
+    |> Enum.all?(&Electric.Features.enabled?/1)
+  end
+
+  def command_list(%SatPerms.DDLX{} = ddlx) do
+    Stream.concat([ddlx.grants, ddlx.revokes, ddlx.assigns, ddlx.unassigns])
+  end
+
+  @feature_flags %{
+    SatPerms.Grant => :proxy_ddlx_grant,
+    SatPerms.Revoke => :proxy_ddlx_revoke,
+    SatPerms.Assign => :proxy_ddlx_assign,
+    SatPerms.Unassign => :proxy_ddlx_unassign,
+    SatPerms.Sqlite => :proxy_ddlx_sqlite
+  }
+
+  # either we have a specific flag for the command or we fallback to the
+  # default setting for the features module, which is `false`
+  defp feature_flag(%cmd{}) do
+    @feature_flags[cmd] || Electric.Features.default_key()
+  end
+
+  def command_id(%SatPerms.Grant{} = grant) do
+    hash([
+      grant.table,
+      grant.role,
+      grant.scope,
+      grant.privilege
+    ])
+  end
+
+  def command_id(%SatPerms.Revoke{} = revoke) do
+    hash([
+      revoke.table,
+      revoke.role,
+      revoke.scope,
+      revoke.privilege
+    ])
+  end
+
+  def command_id(%SatPerms.Assign{} = assign) do
+    hash([
+      assign.table,
+      assign.user_column,
+      assign.role_column,
+      assign.role_name,
+      assign.scope
+    ])
+  end
+
+  def command_id(%SatPerms.Unassign{} = unassign) do
+    hash([
+      unassign.table,
+      unassign.user_column,
+      unassign.role_column,
+      unassign.role_name,
+      unassign.scope
+    ])
+  end
+
+  defp hash(terms) do
+    terms
+    |> Enum.map(&fingerprint/1)
+    |> Enum.intersperse("\n")
+    |> then(&:crypto.hash(:sha, &1))
+    |> Base.encode32(case: :lower, padding: false)
+  end
+
+  defp fingerprint(nil) do
+    <<0>>
+  end
+
+  defp fingerprint(string) when is_binary(string) do
+    string
+  end
+
+  defp fingerprint(%SatPerms.Table{} = table) do
+    [table.schema, ".", table.name]
+  end
+
+  defp fingerprint(%SatPerms.RoleName{role: {:predefined, :AUTHENTICATED}}) do
+    "__electric__.__authenticated__"
+  end
+
+  defp fingerprint(%SatPerms.RoleName{role: {:predefined, :ANYONE}}) do
+    "__electric__.__anyone__"
+  end
 
-  @spec table_name(t()) :: String.t() | {String.t(), String.t()}
-  def table_name(command)
+  defp fingerprint(%SatPerms.RoleName{role: {:application, role}}) do
+    role
+  end
 
-  @spec tag(t()) :: String.t()
-  def tag(command)
+  defp fingerprint(priv) when priv in [:SELECT, :INSERT, :UPDATE, :DELETE] do
+    to_string(priv)
+  end
 
-  @spec to_protobuf(t()) :: [Electric.Satellite.Protobuf.perms_msg()] | []
-  def to_protobuf(command)
+  defimpl Electric.DDLX.Command.PgSQL do
+    def to_sql(%Electric.DDLX.Command{cmds: cmds}) do
+      PgSQL.to_sql(cmds)
+    end
+  end
 end
 
-defimpl Electric.DDLX.Command, for: List do
-  def pg_sql(commands) do
-    Enum.flat_map(commands, &Electric.DDLX.Command.pg_sql/1)
+defimpl Electric.DDLX.Command.PgSQL, for: SatPerms.DDLX do
+  alias Electric.Postgres.Extension
+
+  def to_sql(%SatPerms.DDLX{} = ddlx) do
+    Enum.concat([
+      serialise_ddlx(ddlx),
+      ddlx
+      |> Electric.DDLX.Command.command_list()
+      |> Enum.flat_map(&Electric.DDLX.Command.PgSQL.to_sql/1)
+    ])
   end
 
-  def table_name([cmd]) do
-    Electric.DDLX.Command.table_name(cmd)
+  defp serialise_ddlx(ddlx) do
+    encoded = Protox.encode!(ddlx) |> IO.iodata_to_binary() |> Base.encode16()
+
+    [
+      "INSERT INTO #{Extension.ddlx_table()} (ddlx) VALUES ('\\x#{encoded}'::bytea);"
+    ]
   end
+end
+
+defimpl Electric.DDLX.Command.PgSQL, for: SatPerms.Grant do
+  def to_sql(%SatPerms.Grant{} = _grant) do
+    []
+  end
+end
 
-  def tag([cmd | _commands]) do
-    Electric.DDLX.Command.tag(cmd)
+defimpl Electric.DDLX.Command.PgSQL, for: SatPerms.Revoke do
+  def to_sql(%SatPerms.Revoke{} = _revoke) do
+    []
   end
+end
+
+defimpl Electric.DDLX.Command.PgSQL, for: SatPerms.Assign do
+  import Electric.DDLX.Command.Common
+
+  def to_sql(%SatPerms.Assign{} = assign) do
+    id = Electric.DDLX.Command.command_id(assign)
+
+    [
+      """
+      CALL electric.assign(
+        assignment_id => #{sql_repr(id)},
+        assign_table_full_name => #{sql_repr(assign.table)},
+        scope => #{sql_repr(assign.scope)},
+        user_column_name => #{sql_repr(assign.user_column)},
+        role_name_string => #{sql_repr(assign.role_name)},
+        role_column_name => #{sql_repr(assign.role_column)},
+        if_fn => #{sql_repr(assign.if)}
+      );
+      """
+    ]
+  end
+end
+
+defimpl Electric.DDLX.Command.PgSQL, for: SatPerms.Unassign do
+  import Electric.DDLX.Command.Common
+
+  def to_sql(%SatPerms.Unassign{} = unassign) do
+    id = Electric.DDLX.Command.command_id(unassign)
+
+    [
+      """
+      CALL electric.unassign(
+        assignment_id => #{sql_repr(id)},
+        assign_table_full_name => #{sql_repr(unassign.table)},
+        scope => #{sql_repr(unassign.scope)},
+        user_column_name => #{sql_repr(unassign.user_column)},
+        role_name_string => #{sql_repr(unassign.role_name)},
+        role_column_name => #{sql_repr(unassign.role_column)}
+      );
+      """
+    ]
+  end
+end
 
-  def to_protobuf(cmds) do
-    Enum.flat_map(cmds, &Electric.DDLX.Command.to_protobuf/1)
+defimpl Electric.DDLX.Command.PgSQL, for: SatPerms.Sqlite do
+  def to_sql(%SatPerms.Sqlite{stmt: stmt}) when is_binary(stmt) do
+    [
+      """
+      CALL electric.sqlite(sql => $sqlite$#{stmt}$sqlite$);
+      """
+    ]
   end
 end

components/electric/lib/electric/ddlx/command/assign.ex

@@ -1,27 +1,10 @@
 defmodule Electric.DDLX.Command.Assign do
   alias Electric.DDLX.Command
+  alias Electric.Satellite.SatPerms
 
   import Electric.DDLX.Parser.Build
 
-  @type t() :: %__MODULE__{
-          table_name: String.t(),
-          user_column: String.t(),
-          scope: String.t(),
-          role_name: String.t(),
-          role_column: String.t(),
-          if_statement: String.t()
-        }
-
-  defstruct [
-    :table_name,
-    :user_column,
-    :scope,
-    :role_name,
-    :role_column,
-    :if_statement
-  ]
-
-  def build(params, opts) do
+  def build(params, opts, ddlx) do
     with {:ok, user_table_schema} <- fetch_attr(params, :user_table_schema, default_schema(opts)),
          {:ok, user_table_name} <- fetch_attr(params, :user_table_name),
          {:ok, user_column} <- fetch_attr(params, :user_table_column),
@@ -37,59 +20,24 @@ defmodule Electric.DDLX.Command.Assign do
 
       attrs = Enum.reduce([scope_attrs, user_attrs, role_attrs], [], &Keyword.merge/2)
 
-      {:ok, struct(__MODULE__, attrs)}
-    end
-  end
-
-  defimpl Command do
-    alias Electric.Satellite.SatPerms, as: P
-
-    import Electric.DDLX.Command.Common
-
-    def pg_sql(assign) do
-      [
-        """
-        CALL electric.assign(
-          assign_table_full_name => #{sql_repr(assign.table_name)},
-          scope => #{sql_repr(assign.scope)},
-          user_column_name => #{sql_repr(assign.user_column)},
-          role_name_string => #{sql_repr(assign.role_name)},
-          role_column_name => #{sql_repr(assign.role_column)},
-          if_fn => #{sql_repr(assign.if_statement)}
-        );
-        """
-      ]
-    end
-
-    def table_name(%{table_name: table_name}) do
-      table_name
-    end
-
-    def tag(_a), do: "ELECTRIC ASSIGN"
-
-    def to_protobuf(assign) do
-      %{table_name: {table_schema, table_name}} = assign
-
-      scope =
-        case assign do
-          %{scope: {scope_schema, scope_name}} ->
-            %P.Table{schema: scope_schema, name: scope_name}
-
-          %{scope: nil} ->
-            nil
-        end
-
-      [
-        %P.Assign{
-          # id: assign.id,
-          table: %P.Table{schema: table_schema, name: table_name},
-          user_column: assign.user_column,
-          role_column: assign.role_column,
-          role_name: assign.role_name,
-          scope: scope,
-          if: assign.if_statement
-        }
-      ]
+      {:ok,
+       %Command{
+         cmds: %SatPerms.DDLX{
+           assigns: [
+             %SatPerms.Assign{
+               table: pb_table(attrs[:table_name]),
+               user_column: attrs[:user_column],
+               role_column: attrs[:role_column],
+               role_name: attrs[:role_name],
+               scope: pb_scope(attrs[:scope]),
+               if: attrs[:if_statement]
+             }
+           ]
+         },
+         stmt: ddlx,
+         tables: [attrs[:table_name]],
+         tag: "ELECTRIC ASSIGN"
+       }}
     end
   end
 end

components/electric/lib/electric/ddlx/command/common.ex

@@ -1,6 +1,8 @@
 defmodule Electric.DDLX.Command.Common do
+  alias Electric.Satellite.SatPerms
+
   def sql_repr(nil) do
-    "null"
+    "NULL"
   end
 
   def sql_repr(value) when is_binary(value) do
@@ -24,6 +26,10 @@ defmodule Electric.DDLX.Command.Common do
     ~s['"#{schema}"."#{table}"']
   end
 
+  def sql_repr(%SatPerms.Table{schema: schema, name: name}) do
+    sql_repr({schema, name})
+  end
+
   defp escape_quotes(value) do
     :binary.replace(value, "'", "''", [:global])
   end