electric.ex

defmodule Phoenix.Sync.Electric do
  @moduledoc """
  A `Plug.Router` and `Phoenix.Router` compatible Plug handler that allows
  you to mount the full Electric shape api into your application.

  Unlike `Phoenix.Sync.Router.sync/2` this allows your app to serve
  shapes defined by `table` parameters, much like the Electric application.

  The advantage is that you're free to put your own authentication and
  authorization Plugs in front of this endpoint, integrating the auth for
  your shapes API with the rest of your app.

  ## Configuration

  Before configuring your router, you must install and configure the
  `:phoenix_sync` application.

  See the documentation for [embedding electric](readme.html#installation-and-configuration) for
  details on embedding Electric into your Elixir application.

  ## Plug Integration

  Mount this Plug into your router using `Plug.Router.forward/2`.

      defmodule MyRouter do
        use Plug.Router, copy_opts_to_assign: :config
        use Phoenix.Sync.Electric

        plug :match
        plug :dispatch

        forward "/shapes",
          to: Phoenix.Sync.Electric,
          init_opts: [opts_in_assign: :config]
      end

  You **must** configure your `Plug.Router` with `copy_opts_to_assign` and
  pass the key you configure here (in this case `:config`) to the
  `Phoenix.Sync.Electric` plug in it's `init_opts`.

  In your application, build your Electric configuration using
  `Phoenix.Sync.plug_opts()` and pass the result to your router as
  `phoenix_sync`:

      # in application.ex
      def start(_type, _args) do
        children = [
          {Bandit, plug: {MyRouter, phoenix_sync: Phoenix.Sync.plug_opts()}, port: 4000}
        ]

        Supervisor.start_link(children, strategy: :one_for_one, name: MyApp.Supervisor)
      end

  ## Phoenix Integration

  Use `Phoenix.Router.forward/2` in your router:

      defmodule MyAppWeb.Router do
        use Phoenix.Router

        pipeline :shapes do
          # your authz plugs
        end

        scope "/shapes" do
          pipe_through [:shapes]

          forward "/", Phoenix.Sync.Electric
        end
      end

  As for the Plug integration, include the configuration at runtime
  within the `Application.start/2` callback.

      # in application.ex
      def start(_type, _args) do
        children = [
          # ...
          {MyAppWeb.Endpoint, phoenix_sync: Phoenix.Sync.plug_opts()}
        ]

        Supervisor.start_link(children, strategy: :one_for_one, name: MyApp.Supervisor)
      end

  """

  import Phoenix.Sync.Application, only: [fetch_with_error: 2]

  require Logger

  @behaviour Phoenix.Sync.Adapter
  @behaviour Plug

  @valid_modes [:http, :embedded, :sandbox, :disabled]
  @client_valid_modes @valid_modes -- [:disabled]
  @electric_available? Code.ensure_loaded?(Electric.Application)

  defmacro __using__(_opts \\ []) do
    Phoenix.Sync.Plug.Utils.env!(__CALLER__)

    quote do
      Phoenix.Sync.Plug.Utils.opts_in_assign!(
        [],
        __MODULE__,
        Phoenix.Sync.Plug.Shapes
      )
    end
  end

  @doc false
  @impl Plug
  def init(opts), do: Map.new(opts)

  @doc false
  @impl Plug
  def call(%{private: %{phoenix_endpoint: endpoint}} = conn, _config) do
    api = endpoint.config(:phoenix_sync)

    serve_api(conn, api)
  end

  def call(conn, %{opts_in_assign: key}) do
    api =
      get_in(conn.assigns, [key, :phoenix_sync]) ||
        raise "Unable to retrieve the Electric API configuration from the assigns"

    serve_api(conn, api)
  end

  @doc false
  def serve_api(conn, api) do
    conn = Plug.Conn.fetch_query_params(conn)

    Phoenix.Sync.Adapter.PlugApi.call(api, conn, conn.params)
  end

  @doc false
  def valid_modes, do: @valid_modes

  @doc false
  @impl Phoenix.Sync.Adapter
  def children(env, opts) do
    {mode, electric_opts} =
      opts
      |> set_environment_defaults(env)
      |> electric_opts(env)

    case mode do
      :disabled ->
        {:ok, []}

      :sandbox ->
        if Phoenix.Sync.sandbox_enabled?() do
          {:ok, [Phoenix.Sync.Sandbox]}
        else
          {:error,
           "Sandbox mode is only available if `:ecto_sql` and `:electric` are listed as dependencies"}
        end

      mode when mode in @valid_modes ->
        embedded_children(env, mode, electric_opts)

      invalid_mode ->
        {:error,
         "Invalid mode `#{inspect(invalid_mode)}`. Valid modes are: #{Enum.map_join(@valid_modes, " or ", &"`:#{&1}`")}"}
    end
  end

  @doc false
  @impl Phoenix.Sync.Adapter
  def plug_opts(env, opts) do
    {mode, electric_opts} =
      opts
      |> set_environment_defaults(env)
      |> electric_opts(env)

    # don't need to validate the mode here -- it will have already been
    # validated by children/0 which is run at phoenix_sync startup before the
    # plug opts call even comes through
    case mode do
      :disabled ->
        []

      mode when mode in @valid_modes ->
        plug_opts(env, mode, electric_opts)

      mode ->
        raise ArgumentError,
          message:
            "Invalid `mode` for phoenix_sync: #{inspect(mode)}. Valid modes are: #{Enum.map_join(@valid_modes, " or ", &"`:#{&1}`")}"
    end
  end

  @doc false
  @impl Phoenix.Sync.Adapter
  def client(env, opts) do
    {mode, electric_opts} =
      opts |> set_environment_defaults(env) |> electric_opts(env)

    case mode do
      mode when mode in @client_valid_modes ->
        env
        |> core_configuration(electric_opts)
        |> configure_client(mode)

      invalid_mode ->
        {:error, "Cannot configure client for mode #{inspect(invalid_mode)}"}
    end
  end

  # if we want to set up per-run configuration, and avoid weird state errors in
  # dev and test, then we have to write them to the application config, because
  # `children/2`, `client/2` and `plug_opts/2` need to have consistent
  # configuration values.
  defp set_environment_defaults(opts, :test) do
    opts
    |> set_persistent_config(:stack_id, fn ->
      "electric-stack#{System.monotonic_time()}"
    end)
    |> set_persistent_config(:replication_stream_id, fn ->
      String.replace("phoenix_sync#{System.monotonic_time()}", "-", "_")
    end)
    |> set_persistent_config(:replication_slot_temporary?, true)
  end

  defp set_environment_defaults(opts, :dev) do
    opts
    |> set_environment_defaults(:prod)
    |> set_persistent_config(:storage_dir, fn ->
      Path.join([System.tmp_dir!(), "phoenix-sync#{System.monotonic_time()}"])
    end)
  end

  defp set_environment_defaults(opts, _env) do
    opts
    |> set_persistent_config(:stack_id, "electric-embedded")
  end

  defp set_persistent_config(opts, key, value_fun) when is_function(value_fun) do
    Keyword.put_new_lazy(opts, key, fn ->
      value = value_fun.()
      Application.put_env(:phoenix_sync, key, value)
      value
    end)
  end

  defp set_persistent_config(opts, key, value) do
    set_persistent_config(opts, key, fn -> value end)
  end

  @doc false
  def electric_available? do
    @electric_available?
  end

  defp electric_opts(opts, env) do
    Keyword.pop_lazy(opts, :mode, fn ->
      default_mode(env)
    end)
  end

  defp default_mode(:test) do
    :disabled
  end

  if @electric_available? do
    defp default_mode(_env) do
      Logger.warning([
        "missing mode configuration for :phoenix_sync. Electric is installed so assuming `embedded` mode"
      ])

      :embedded
    end
  else
    defp default_mode(_env) do
      Logger.warning("No `:mode` configuration for :phoenix_sync, assuming `:disabled`")

      :disabled
    end
  end

  defp plug_opts(_env, :http, opts) do
    case http_mode_plug_opts(opts) do
      {:ok, config} -> config
      {:error, reason} -> raise ArgumentError, message: reason
    end
  end

  if @electric_available? do
    defp plug_opts(env, :embedded, electric_opts) do
      env
      |> core_configuration(electric_opts)
      |> Electric.Application.api_plug_opts()
      |> Keyword.fetch!(:api)
    end

    if Phoenix.Sync.sandbox_enabled?() do
      defp plug_opts(_env, :sandbox, _electric_opts) do
        Phoenix.Sync.Sandbox.APIAdapter.new()
      end
    else
      defp plug_opts(_env, :sandbox, _electric_opts) do
        raise ArgumentError,
          message:
            "phoenix_sync configured in `mode: :sandbox` but Ecto.SQL not installed. Please add `:ecto_sql` to your dependencies or use `:http` mode."
      end
    end
  else
    defp plug_opts(_env, :embedded, _electric_opts) do
      raise ArgumentError,
        message:
          "phoenix_sync configured in `mode: :embedded` but electric not installed. Please add `:electric` to your dependencies or use `:http` mode."
    end

    defp plug_opts(_env, :sandbox, _electric_opts) do
      raise ArgumentError,
        message:
          "phoenix_sync configured in `mode: :sandbox` but electric not installed. Please add `:electric` to your dependencies or use `:http` mode."
    end
  end

  defp embedded_children(_env, :disabled, _opts) do
    {:ok, []}
  end

  defp embedded_children(env, mode, opts) do
    electric_children(env, mode, opts)
  end

  defp electric_children(env, mode, opts) do
    case validate_database_config(env, mode, opts) do
      {:start, db_config_fun, message} ->
        start_embedded(env, mode, db_config_fun, message)

      :ignore ->
        {:ok, []}

      {:error, _} = error ->
        error
    end
  end

  if @electric_available? do
    defp start_embedded(env, mode, db_config_fun, message) do
      db_config = db_config_fun.()

      electric_config = core_configuration(env, db_config)

      Logger.info(message)

      http_server =
        case mode do
          :http -> electric_api_server(electric_config)
          _ -> []
        end

      {:ok,
       [
         {Electric.StackSupervisor, Electric.Application.configuration(electric_config)}
         | http_server
       ]}
    end

    defp electric_api_server(opts) do
      config = electric_http_config(opts)

      cond do
        Code.ensure_loaded?(Bandit) ->
          Electric.Application.api_server(Bandit, config)

        Code.ensure_loaded?(Plug.Cowboy) ->
          Electric.Application.api_server(Plug.Cowboy, config)

        true ->
          raise RuntimeError,
            message: "No HTTP server found. Please install either Bandit or Plug.Cowboy"
      end
    end

    defp electric_http_config(opts) do
      case Keyword.fetch(opts, :http) do
        {:ok, http_opts} ->
          opts
          |> then(fn o ->
            if(port = http_opts[:port], do: Keyword.put(o, :service_port, port), else: o)
          end)

        :error ->
          opts
      end
      # TODO: remove this once https://github.com/electric-sql/electric/pull/2863
      # is released
      |> Keyword.put_new(:http_api_num_acceptors, nil)
    end
  else
    defp start_embedded(_env, _mode, _db_config_fun, _message) do
      {:error,
       "Electric configured to start in embedded mode but :electric dependency not available"}
    end
  end

  defp stack_id(opts) do
    Keyword.put_new(opts, :stack_id, "electric-embedded")
  end

  defp core_configuration(env, opts) do
    opts
    |> env_defaults(env)
    |> stack_id()
    |> overrides()
  end

  defp env_defaults(opts, :dev) do
    # can't use new tmp dir for every run in dev because the storage
    # path must remain consistent between invocations of children(), plug_opts()
    # and client()...
    # if we want to use emphemeral dir for dev storage then we have to persist
    # the storage_dir into the application config.
    opts
    |> Keyword.put_new(:send_cache_headers?, false)
  end

  defp env_defaults(opts, :test) do
    stack_id = "electric-stack"

    opts = Keyword.put_new(opts, :stack_id, stack_id)

    opts
    |> Keyword.put_new(
      :storage,
      {Electric.ShapeCache.InMemoryStorage,
       table_base_name: :"electric-storage#{opts[:stack_id]}", stack_id: opts[:stack_id]}
    )
    |> Keyword.put_new(
      :persistent_kv,
      {Electric.PersistentKV.Memory, :new!, []}
    )
  end

  defp env_defaults(opts, _) do
    opts
  end

  defp overrides(opts) do
    opts
    |> Keyword.put_new(:stack_ready_timeout, 5_000)
  end

  # Returns a function to generate the config so that we can
  # centralise the test for the existence of electric.
  # Need this because the convert_repo_config/1 function needs Electric
  # installed too
  defp validate_database_config(_env, mode, opts) do
    case Keyword.pop(opts, :repo, nil) do
      {nil, opts} ->
        case {Keyword.get(opts, :connection_opts),
              Keyword.get(opts, :replication_connection_opts)} do
          {[_ | _] = connection_opts, _} ->
            opts =
              opts
              |> Keyword.delete(:connection_opts)
              |> Keyword.put(:replication_connection_opts, connection_opts)

            {:start, fn -> opts end, start_message(connection_opts)}

          {nil, [_ | _] = connection_opts} ->
            {:start, fn -> opts end, start_message(connection_opts)}

          {nil, nil} ->
            case mode do
              :embedded ->
                {:error,
                 "No database configuration available. Include either a `repo` or a `connection_opts` in the phoenix_sync `electric` config"}

              :http ->
                :ignore
            end
        end

      {repo, opts} when is_atom(repo) ->
        if Code.ensure_loaded?(repo) && function_exported?(repo, :config, 0) do
          repo_config = apply(repo, :config, [])

          {:start,
           fn ->
             connection_opts = convert_repo_config(repo_config)

             Keyword.put(opts, :replication_connection_opts, connection_opts)
           end, "Starting Electric replication stream using #{repo} configuration"}
        else
          {:error, "#{inspect(repo)} is not a valid Ecto.Repo module"}
        end
    end
  end

  defp start_message(connection_opts) do
    "Starting Electric replication stream from postgresql://#{connection_opts[:host]}:#{connection_opts[:port] || 5432}/#{connection_opts[:database]}"
  end

  if @electric_available? do
    defp convert_repo_config(repo_config) do
      expected_keys = Electric.connection_opts_schema() |> Keyword.keys()

      ssl_opts =
        case Keyword.get(repo_config, :ssl, nil) do
          off when off in [nil, false] -> [sslmode: :disable]
          true -> [sslmode: :require]
          _opts -> []
        end

      tcp_opts =
        if :inet6 in Keyword.get(repo_config, :socket_options, []),
          do: [ipv6: true],
          else: []

      repo_config
      |> Keyword.take(expected_keys)
      |> Keyword.merge(ssl_opts)
      |> Keyword.merge(tcp_opts)
      |> Keyword.put_new(:port, 5432)
    end
  else
    defp convert_repo_config(_repo_config) do
      []
    end
  end

  defp http_mode_plug_opts(electric_config) do
    with {:ok, client} <- configure_client(electric_config, :http) do
      # don't decode the body - just pass it directly
      client = %{client | fetch: {Electric.Client.Fetch.HTTP, [request: [raw: true]]}}
      {:ok, %Phoenix.Sync.Electric.ClientAdapter{client: client}}
    end
  end

  if @electric_available? do
    defp configure_client(opts, :embedded) do
      Electric.Client.embedded(opts)
    end

    # with a sandbox config, we're basically abandoning the idea of connecting
    # to a real instance -- I think that's reasonable. The overhead of a real
    # electric consuming a real replication stream is way too high for a simple
    # consumer of streams
    if Phoenix.Sync.sandbox_enabled?() do
      defp configure_client(_opts, :sandbox) do
        Phoenix.Sync.Sandbox.client()
      end
    else
      defp configure_client(_opts, :sandbox) do
        {:error, "sandbox mode is only available if Ecto.SQL is installed"}
      end
    end
  else
    defp configure_client(_opts, :embedded) do
      {:error, "electric not installed, unable to created embedded client"}
    end

    defp configure_client(_opts, :sandbox) do
      {:error, "electric not installed, unable to created sandbox client"}
    end
  end

  defp configure_client(electric_config, :http) do
    with {:ok, url} <- fetch_with_error(electric_config, :url),
         credential_params = electric_config |> Keyword.get(:credentials, []) |> Map.new(),
         extra_params = electric_config |> Keyword.get(:params, []) |> Map.new(),
         params = Map.merge(extra_params, credential_params) do
      Electric.Client.new(
        base_url: url,
        params: params
      )
    end
  end

  @doc false
  def api_predefined_shape(conn, api, shape, response_fun) when is_function(response_fun, 2) do
    case Phoenix.Sync.Adapter.PlugApi.predefined_shape(api, shape) do
      {:ok, shape_api} ->
        # response_fun should return conn
        response_fun.(conn, shape_api)

      # Only the embedded api will ever return an error from predefined_shape/2
      # when the stack isn't ready (or the params are invalid, e.g. bad table).
      # The client adapter just configures the client with the shape
      # parameters, which can't error.
      {:error, response} ->
        conn
        |> Plug.Conn.send_resp(response.status, Enum.into(response.body, []))
    end
  end

  @subset_body_keys ~w(where order_by limit offset params where_expr order_by_expr)
  # In flattened POST params, root offset may be the stream cursor, so treat offset
  # as subset-only only when we can reliably separate query/body params.
  @subset_body_keys_without_offset @subset_body_keys -- ["offset"]

  @doc false
  def normalize_subset_params(params, method \\ "GET")

  def normalize_subset_params(%Plug.Conn{} = conn, params) when is_map(params) do
    case conn.method do
      "POST" ->
        query_params = map_or_empty(conn.query_params) |> normalize_prefixed_subset_params()
        body_params = map_or_empty(conn.body_params)
        path_params = map_or_empty(conn.path_params)

        # Match sync-service behavior by nesting subset body keys separately
        # from query params, preserving stream offset/handle from query/path.
        if map_size(query_params) == 0 and map_size(body_params) == 0 do
          normalize_subset_params(params, conn.method)
        else
          query_params
          |> merge_subset_body_params(body_params)
          |> Map.merge(path_params)
        end

      _ ->
        normalize_subset_params(params, conn.method)
    end
  end

  def normalize_subset_params(params, method) when is_map(params) do
    params
    |> normalize_prefixed_subset_params()
    |> maybe_normalize_subset_body_params(method)
  end

  defp normalize_prefixed_subset_params(params) when is_map(params) do
    {subset_prefixed_params, rest} =
      Enum.reduce(params, {%{}, %{}}, fn {key, value}, {subset_acc, rest_acc} ->
        case subset_param_name(key) do
          {:ok, subset_key} ->
            {Map.put(subset_acc, subset_key, value), rest_acc}

          :error ->
            {subset_acc, Map.put(rest_acc, key, value)}
        end
      end)

    if map_size(subset_prefixed_params) == 0 do
      params
    else
      existing_subset = existing_subset(rest)

      rest
      |> Map.drop(["subset", :subset])
      |> Map.put("subset", Map.merge(existing_subset, subset_prefixed_params))
    end
  end

  # POST subset snapshots send params in the body as plain subset keys
  # (where/params/limit/offset/order_by), which need nesting under "subset".
  defp maybe_normalize_subset_body_params(params, method)
       when method in ["POST", :post] and is_map(params) do
    {subset_body_params, rest} =
      Enum.reduce(params, {%{}, %{}}, fn {key, value}, {subset_acc, rest_acc} ->
        case subset_body_param_name(key, preserve_stream_offset?: true) do
          {:ok, subset_key} ->
            {Map.put(subset_acc, subset_key, value), rest_acc}

          :error ->
            {subset_acc, Map.put(rest_acc, key, value)}
        end
      end)

    if map_size(subset_body_params) == 0 do
      params
    else
      existing_subset = existing_subset(rest)

      rest
      |> Map.drop(["subset", :subset])
      |> Map.put("subset", Map.merge(existing_subset, subset_body_params))
    end
  end

  defp maybe_normalize_subset_body_params(params, _method), do: params

  defp subset_param_name(key) when is_binary(key) do
    if String.starts_with?(key, "subset__") do
      {:ok, String.replace_prefix(key, "subset__", "")}
    else
      :error
    end
  end

  defp subset_param_name(key) when is_atom(key) do
    key
    |> Atom.to_string()
    |> subset_param_name()
  end

  defp subset_param_name(_), do: :error

  defp subset_body_param_name(key, opts \\ [])

  defp subset_body_param_name(key, opts) when is_binary(key) do
    keys =
      if opts[:preserve_stream_offset?],
        do: @subset_body_keys_without_offset,
        else: @subset_body_keys

    if key in keys, do: {:ok, key}, else: :error
  end

  defp subset_body_param_name(key, opts) when is_atom(key) do
    key
    |> Atom.to_string()
    |> subset_body_param_name(opts)
  end

  defp subset_body_param_name(_, _opts), do: :error

  defp existing_subset(params) when is_map(params) do
    params
    |> Map.take(["subset", :subset])
    |> Map.values()
    |> Enum.filter(&is_map/1)
    |> Enum.reduce(%{}, &Map.merge(&2, &1))
  end

  defp merge_subset_body_params(query_params, body_params) when map_size(body_params) == 0 do
    query_params
  end

  defp merge_subset_body_params(query_params, body_params) when is_map(body_params) do
    case existing_subset(body_params) do
      subset when map_size(subset) > 0 ->
        existing_subset = existing_subset(query_params)

        query_params
        |> Map.merge(body_params)
        |> Map.drop([:subset, "subset"])
        |> Map.put("subset", Map.merge(existing_subset, subset))

      _ ->
        {subset_params, other_params} =
          Enum.reduce(body_params, {%{}, %{}}, fn {key, value}, {subset_acc, rest_acc} ->
            case subset_body_param_name(key) do
              {:ok, subset_key} ->
                {Map.put(subset_acc, subset_key, value), rest_acc}

              :error ->
                {subset_acc, Map.put(rest_acc, key, value)}
            end
          end)

        if map_size(subset_params) > 0 do
          existing_subset = existing_subset(query_params)

          query_params
          |> Map.merge(other_params)
          |> Map.put("subset", Map.merge(existing_subset, subset_params))
        else
          Map.merge(query_params, body_params)
        end
    end
  end

  defp map_or_empty(%Plug.Conn.Unfetched{}), do: %{}
  defp map_or_empty(map) when is_map(map), do: map
  defp map_or_empty(_), do: %{}

  @json Phoenix.Sync.json_library()

  @doc false
  def map_response_body(body, nil) do
    body
  end

  # empty body is a valid response but not valid JSON
  def map_response_body("", _mapper) do
    ""
  end

  def map_response_body(body, mapper) when is_binary(body) and is_function(mapper, 1) do
    body
    |> @json.decode!()
    |> map_response_body(mapper)
    |> then(fn item -> [@json.encode_to_iodata!(item)] end)
  end

  def map_response_body(msgs, mapper) when is_list(msgs) and is_function(mapper, 1) do
    msgs
    |> Enum.flat_map(fn
      %{"key" => _key, "headers" => _, "value" => _} = msg ->
        mapper.(msg)

      control ->
        [control]
    end)
  end

  def map_response_body(msgs, _mapper) do
    msgs
  end

  if @electric_available? do
    @doc false
    # for the embedded api we need to make sure that the response stream is consumed
    # in the same process that made the request, in order for cleanups to happen
    # so we have to enumerate the body stream immediately before passing the response
    # to any other process...
    def consume_response_stream(%Electric.Shapes.Api.Response{} = response) do
      Map.update!(response, :body, &do_consume_stream(&1))
    end

    defp do_consume_stream(body) do
      Enum.to_list(body)
    end
  end
end

if Code.ensure_loaded?(Electric.Shapes.Api) do
  Code.ensure_loaded(Phoenix.Sync.Electric.ApiAdapter)

  defimpl Phoenix.Sync.Adapter.PlugApi, for: Electric.Shapes.Api do
    alias Electric.Shapes

    alias Phoenix.Sync.PredefinedShape
    alias Phoenix.Sync.Electric.ApiAdapter

    def predefined_shape(api, %PredefinedShape{} = shape) do
      ApiAdapter.new(api, shape)
    end

    def call(api, %{method: "GET"} = conn, params) do
      params = Phoenix.Sync.Electric.normalize_subset_params(conn, params)

      case Shapes.Api.validate(api, params) do
        {:ok, request} ->
          conn
          |> content_type()
          |> Plug.Conn.assign(:request, request)
          |> Shapes.Api.serve_shape_response(request)

        {:error, response} ->
          conn
          |> content_type()
          |> Shapes.Api.Response.send(response)
          |> Plug.Conn.halt()
      end
    end

    def call(api, %{method: "DELETE"} = conn, params) do
      case Shapes.Api.validate_for_delete(api, params) do
        {:ok, request} ->
          conn
          |> content_type()
          |> Plug.Conn.assign(:request, request)
          |> Shapes.Api.delete_shape(request)

        {:error, response} ->
          conn
          |> content_type()
          |> Shapes.Api.Response.send(response)
          |> Plug.Conn.halt()
      end
    end

    def call(_api, %{method: "OPTIONS"} = conn, _params) do
      Shapes.Api.options(conn)
    end

    def response(api, conn, params) do
      params = Phoenix.Sync.Electric.normalize_subset_params(conn, params)

      case Shapes.Api.validate(api, params) do
        {:ok, request} ->
          {
            request,
            Shapes.Api.serve_shape_response(request)
            |> Phoenix.Sync.Electric.consume_response_stream()
          }

        {:error, response} ->
          {nil, response}
      end
    end

    def send_response(_api, conn, {request, response}) do
      conn
      |> content_type()
      |> Plug.Conn.assign(:request, request)
      |> Plug.Conn.assign(:response, response)
      |> Shapes.Api.Response.send(response)
    end

    defp content_type(conn) do
      Plug.Conn.put_resp_content_type(conn, "application/json")
    end
  end
end