sync-service: improve shapes api encapsulation (#2376)

- Api.predefined_shape/2 allows for configuring an endpoint with a
pre-defined shape
- All http logic is now encapsulated by the api, including not-modified
handling and cache-control headers

Author: magnetised

.changeset/seven-guests-appear.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Allow for accessing via an api that serves only a pre-configured shape, move all http logic into api

packages/sync-service/lib/electric/plug/serve_shape_plug.ex

@@ -5,30 +5,19 @@ defmodule Electric.Plug.ServeShapePlug do
   # The halt/1 function is redefined further down below
   import Plug.Conn, except: [halt: 1]
 
-  alias Electric.Plug.Utils
   alias Electric.Shapes.Api
-  alias Electric.Replication.LogOffset
   alias Electric.Telemetry.OpenTelemetry
   alias Plug.Conn
 
   require Logger
 
-  defguardp is_live_request(conn) when conn.assigns.request.params.live
-
-  # Aliasing for pattern matching
-  @before_all_offset LogOffset.before_all()
-
   plug :fetch_query_params
 
   # start_telemetry_span needs to always be the first plug after fetching query params.
   plug :start_telemetry_span
   plug :put_resp_content_type, "application/json"
 
   plug :validate_request
-  plug :put_schema_header
-  plug :put_resp_cache_headers
-  plug :generate_etag
-  plug :validate_and_put_etag
   plug :serve_shape_log
 
   # end_telemetry_span needs to always be the last plug here.
@@ -54,98 +43,8 @@ defmodule Electric.Plug.ServeShapePlug do
     end
   end
 
-  # Only adds schema header when not in live mode
-  defp put_schema_header(conn, _) when not is_live_request(conn) do
-    %{assigns: %{request: request}} = conn
-    schema = Api.schema(request) |> Jason.encode!()
-    put_resp_header(conn, "electric-schema", schema)
-  end
-
-  defp put_schema_header(conn, _), do: conn
-
-  defp generate_etag(%Conn{} = conn, _) do
-    %{assigns: %{request: request}} = conn
-
-    %{
-      handle: active_shape_handle,
-      offset: chunk_end_offset
-    } = request.response
-
-    conn
-    |> assign(
-      :etag,
-      "#{active_shape_handle}:#{request.params.offset}:#{chunk_end_offset}"
-    )
-  end
-
-  defp validate_and_put_etag(%Conn{} = conn, _) do
-    %{assigns: %{request: request}} = conn
-
-    if_none_match =
-      get_req_header(conn, "if-none-match")
-      |> Enum.flat_map(&String.split(&1, ","))
-      |> Enum.map(&String.trim/1)
-      |> Enum.map(&String.trim(&1, <<?">>))
-
-    cond do
-      conn.assigns.etag in if_none_match ->
-        conn
-        |> send_resp(304, "")
-        |> halt()
-
-      not request.params.live ->
-        put_resp_header(conn, "etag", conn.assigns.etag)
-
-      true ->
-        conn
-    end
-  end
-
-  defp put_resp_cache_headers(%Conn{} = conn, _) do
-    %{assigns: %{request: request}} = conn
-
-    case request do
-      # If the offset is -1, set a 1 week max-age, 1 hour s-maxage (shared cache)
-      # and 1 month stale-while-revalidate We want private caches to cache the
-      # initial offset for a long time but for shared caches to frequently
-      # revalidate so they're serving a fairly fresh copy of the initials shape
-      # log.
-      %{params: %{offset: @before_all_offset}} ->
-        conn
-        |> put_resp_header(
-          "cache-control",
-          "public, max-age=604800, s-maxage=3600, stale-while-revalidate=2629746"
-        )
-
-      # For live requests we want short cache lifetimes and to update the live cursor
-      %{params: %{live: true}, api: api} ->
-        conn
-        |> put_resp_header(
-          "cache-control",
-          "public, max-age=5, stale-while-revalidate=5"
-        )
-        |> put_resp_header(
-          "electric-cursor",
-          api.long_poll_timeout
-          |> Utils.get_next_interval_timestamp(conn.query_params["cursor"])
-          |> Integer.to_string()
-        )
-
-      %{params: %{live: false}, api: api} ->
-        conn
-        |> put_resp_header(
-          "cache-control",
-          "public, max-age=#{api.max_age}, stale-while-revalidate=#{api.stale_age}"
-        )
-    end
-  end
-
   defp serve_shape_log(%Conn{assigns: %{request: request}} = conn, _) do
-    response = Api.serve_shape_log(request)
-
-    conn
-    |> assign(:response, response)
-    |> Api.Response.send(response)
+    Api.serve_shape_log(conn, request)
   end
 
   defp open_telemetry_attrs(%Conn{assigns: assigns} = conn) do

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

@@ -13,7 +13,7 @@ defmodule Electric.Shapes.Api do
 
   @options [
     inspector: [type: :mod_arg, required: true],
-    pg_id: [type: :string],
+    pg_id: [type: {:or, [nil, :string]}],
     registry: [type: :atom, required: true],
     shape_cache: [type: :mod_arg, required: true],
     stack_events_registry: [type: :atom, required: true],
@@ -40,6 +40,7 @@ defmodule Electric.Shapes.Api do
     :pg_id,
     :registry,
     :persistent_kv,
+    :shape,
     :shape_cache,
     :stack_events_registry,
     :stack_id,
@@ -105,19 +106,60 @@ defmodule Electric.Shapes.Api do
   end
 
   defp validate_encoder!(%Api{} = api) do
-    Map.update!(api, :encoder, &Electric.Shapes.Api.Encoder.validate!/1)
+    Map.update!(api, :encoder, &Shapes.Api.Encoder.validate!/1)
   end
 
+  shape_schema_options =
+    Keyword.merge(Keyword.drop(Shapes.Shape.schema_options(), [:inspector]),
+      table: [type: :string],
+      schema: [type: :string],
+      namespace: [type: :string]
+    )
+
+  shape_schema = NimbleOptions.new!(shape_schema_options)
+
+  @type shape_opts() :: [unquote(NimbleOptions.option_typespec(shape_schema))]
+
   @doc """
-  Validate the parameters for the request.
+  Create a version of the given configured Api instance that is specific to the
+  given shape.
 
-  Options:
+  This allows you to provide a locked-down version of the API that ignores
+  shape-definition parameters such as `table`, `where` and `columns` and only
+  honours the shape-tailing parameters such as `offset` and `handle`.
+  """
+  @spec predefined_shape(t(), shape_opts()) :: {:ok, t()} | {:error, term()}
+  def predefined_shape(%Api{} = api, shape_params) do
+    with {:ok, params} <- normalise_shape_params(shape_params),
+         opts = Keyword.merge(params, inspector: api.inspector),
+         {:ok, shape} <- Shapes.Shape.new(opts) do
+      {:ok, %{api | shape: shape}}
+    end
+  end
 
-  - `seek: boolean()` - (default: true) once validated should we load the shape's
-    latest offset information.
+  defp normalise_shape_params(params) do
+    case Keyword.fetch(params, :relation) do
+      {:ok, {n, t}} when is_binary(n) and is_binary(t) ->
+        {:ok, params}
+
+      :error ->
+        {table_params, shape_params} = Keyword.split(params, [:table, :namespace, :schema])
+
+        case {table_params[:table], table_params[:namespace] || table_params[:schema]} do
+          {nil, nil} ->
+            {:error, "No relation or table specified"}
+
+          {table, nil} when is_binary(table) ->
+            {:ok, Keyword.put(shape_params, :relation, {"public", table})}
+
+          {table, namespace} ->
+            {:ok, Keyword.put(shape_params, :relation, {namespace, table})}
+        end
+    end
+  end
 
-  - `load: boolean()` - (default: true) validate and optionallly create a shape
-    based on the handle and shape parameters.
+  @doc """
+  Validate the parameters for the request.
   """
   @spec validate(t(), %{(atom() | binary()) => term()}) ::
           {:ok, Request.t()} | {:error, Response.t()}
@@ -139,9 +181,15 @@ defmodule Electric.Shapes.Api do
 
   defp validate_params(api, params) do
     with {:ok, request_params} <- Api.Params.validate(api, params) do
-      request_for_params(api, request_params, %Response{
-        shape_definition: request_params.shape_definition
-      })
+      request_for_params(
+        api,
+        request_params,
+        %Response{
+          api: api,
+          params: request_params,
+          shape_definition: request_params.shape_definition
+        }
+      )
     end
   end
 
@@ -352,6 +400,39 @@ defmodule Electric.Shapes.Api do
     end)
   end
 
+  def serve_shape_log(%Plug.Conn{} = conn, %Request{} = request) do
+    response =
+      case if_not_modified(conn, request) do
+        {:halt, response} ->
+          response
+
+        {:cont, request} ->
+          serve_shape_log(request)
+      end
+
+    conn
+    |> Plug.Conn.assign(:response, response)
+    |> Response.send(response)
+  end
+
+  def if_not_modified(conn, request) do
+    etag = Response.etag(request.response, quote: false)
+
+    if etag in if_none_match(conn) do
+      %{response: response} = Request.update_response(request, &%{&1 | status: 304, body: []})
+      {:halt, response}
+    else
+      {:cont, request}
+    end
+  end
+
+  defp if_none_match(%Plug.Conn{} = conn) do
+    Plug.Conn.get_req_header(conn, "if-none-match")
+    |> Enum.flat_map(&String.split(&1, ","))
+    |> Enum.map(&String.trim/1)
+    |> Enum.map(&String.trim(&1, <<?">>))
+  end
+
   defp validate_serve_usage!(request) do
     case {request.new_changes_pid, self()} do
       {nil, _} ->

packages/sync-service/lib/electric/shapes/api/params.ex

@@ -56,11 +56,11 @@ defmodule Electric.Shapes.Api.Params do
   def validate(%Electric.Shapes.Api{} = api, params) do
     params
     |> cast_params()
-    |> validate_required([:table, :offset])
+    |> validate_required([:offset])
     |> cast_offset()
     |> validate_handle_with_offset()
     |> validate_live_with_offset()
-    |> cast_root_table(inspector: api.inspector)
+    |> cast_root_table(api)
     |> apply_action(:validate)
     |> convert_error(api)
   end
@@ -77,7 +77,7 @@ defmodule Electric.Shapes.Api.Params do
       %{changes: %{table: _table}} = changeset ->
         changeset
         |> validate_required([:table])
-        |> cast_root_table(inspector: api.inspector)
+        |> cast_root_table(api)
         |> apply_action(:validate)
         |> convert_error(api)
 
@@ -151,7 +151,21 @@ defmodule Electric.Shapes.Api.Params do
 
   def cast_root_table(%Ecto.Changeset{valid?: false} = changeset, _), do: changeset
 
-  def cast_root_table(%Ecto.Changeset{} = changeset, opts) do
+  def cast_root_table(%Ecto.Changeset{} = changeset, %Api{shape: nil} = api) do
+    changeset
+    |> validate_required([:table])
+    |> define_shape(api)
+  end
+
+  def cast_root_table(%Ecto.Changeset{} = changeset, %Api{shape: %Shape{} = shape}) do
+    put_change(changeset, :shape_definition, shape)
+  end
+
+  defp define_shape(%Ecto.Changeset{valid?: false} = changeset, _api) do
+    changeset
+  end
+
+  defp define_shape(%Ecto.Changeset{} = changeset, api) do
     table = fetch_change!(changeset, :table)
     where = fetch_field!(changeset, :where)
     columns = get_change(changeset, :columns, nil)
@@ -160,16 +174,14 @@ defmodule Electric.Shapes.Api.Params do
 
     case Shape.new(
            table,
-           opts ++
-             [
-               where: where,
-               columns: columns,
-               replica: replica,
-               storage: %{compaction: if(compaction_enabled?, do: :enabled, else: :disabled)}
-             ]
+           where: where,
+           columns: columns,
+           replica: replica,
+           inspector: api.inspector,
+           storage: %{compaction: if(compaction_enabled?, do: :enabled, else: :disabled)}
          ) do
-      {:ok, result} ->
-        put_change(changeset, :shape_definition, result)
+      {:ok, shape} ->
+        put_change(changeset, :shape_definition, shape)
 
       {:error, {field, reasons}} ->
         Enum.reduce(List.wrap(reasons), changeset, fn