phoenix_sync

defmodule Phoenix.Sync.Client do
  @moduledoc """
  Low level Elixir client. Converts an `Ecto.Query` into an Elixir `Stream`:

  ```elixir
  stream = Phoenix.Sync.Client.stream(Todos.Todo)

  stream =
    Ecto.Query.from(t in Todos.Todo, where: t.completed == false)
    |> Phoenix.Sync.Client.stream()
  ```
  """

  alias Phoenix.Sync.PredefinedShape

  @doc """
  Create a new sync client based on the `:phoenix_sync` configuration.
  """
  def new do
    Phoenix.Sync.Application.config() |> new()
  end

  def new(nil) do
    new()
  end

  @doc """
  Create a sync client using the given options.

  If the integration mode is set to `:embedded` and Electric is installed
  then this will configure the client to retrieve data using the internal
  Elixir APIs.

  For the `:http` mode, then you must also configure a URL specifying an
  Electric API server:

      config :phoenix_sync,
        mode: :http,
        url: "https://api.electric-sql.cloud"

  This client can then generate streams for use in your Elixir applications:

      {:ok, client} = Phoenix.Sync.Client.new()
      stream = Electric.Client.stream(client, Todos.Todo)
      for msg <- stream, do: IO.inspect(msg)

  Alternatively use `stream/1` which wraps this functionality.
  """
  def new(opts) do
    {adapter, env} = Phoenix.Sync.Application.adapter_env(opts)

    apply(adapter, :client, [env, opts])
  end

  @doc """
  Create a new sync client based on the application configuration or raise if
  the config is invalid.

      client = Phoenix.Sync.Client.new!()

  See `new/0`.
  """
  def new! do
    case new() do
      {:ok, client} ->
        client

      {:error, reason} ->
        raise RuntimeError, message: "Invalid client configuration: #{reason}"
    end
  end

  @doc """
  Create a new sync client based on the given opts or raise if
  the config is invalid.

      client = Phoenix.Sync.Client.new!(mode: :embedded)

  See `new/1`.
  """
  def new!(opts) do
    case new(opts) do
      {:ok, client} ->
        client

      {:error, reason} ->
        raise RuntimeError, message: "Invalid client configuration: #{reason}"
    end
  end

  @doc """
  Return a sync stream for the given shape.

  ## Examples

      # stream updates for the Todo schema
      stream = Phoenix.Sync.Client.stream(MyApp.Todos.Todo)

      # stream the results of an ecto query
      stream = Phoenix.Sync.Client.stream(from(t in MyApp.Todos.Todo, where: t.completed == true))

      # create a stream based on a shape definition
      stream = Phoenix.Sync.Client.stream(
        table: "todos",
        where: "completed = false",
        columns: ["id", "title"]
      )

      # once you have a stream, consume it as usual
      Enum.each(stream, &IO.inspect/1)

  ## Ecto vs keyword shapes

  Streams defined using an Ecto query or schema will return data wrapped in
  the appropriate schema struct, with values cast to the appropriate
  Elixir/Ecto types, rather than raw column data in the form `%{"column_name"
  => "column_value"}`.
  """
  @spec stream(Phoenix.Sync.shape_definition(), Electric.Client.stream_options()) :: Enum.t()
  def stream(shape, stream_opts \\ [])

  def stream(table, stream_opts) when is_binary(table) and is_list(stream_opts) do
    stream(Keyword.put(stream_opts, :table, table), [])
  end

  def stream(shape, []) when is_list(shape) do
    {client, shape} = Keyword.pop_lazy(shape, :client, &new!/0)

    {shape, shape_stream_opts} = resolve_shape(shape, [])

    Electric.Client.stream(client, shape, shape_stream_opts)
  end

  def stream(shape, stream_opts) when not is_list(shape) and is_list(stream_opts) do
    {client, stream_opts} = Keyword.pop_lazy(stream_opts, :client, &new!/0)
    {shape, shape_stream_opts} = resolve_shape(shape, stream_opts)

    Electric.Client.stream(client, shape, shape_stream_opts)
  end

  defp resolve_shape(shape, stream_opts) do
    shape
    |> PredefinedShape.new!(stream_opts)
    |> Phoenix.Sync.PredefinedShape.to_stream_params()
  end
end