uma_db_client

defmodule UmaDbClient do
  @moduledoc """
  Elixir gRPC client for UmaDB's DCB (Dynamic Consistency Boundary) service.

  ## Usage

      {:ok, channel} = UmaDbClient.connect("localhost:50051")

      # Append events
      event = UmaDbClient.Builder.event(type: "OrderPlaced", tags: ["order:1"], data: Jason.encode!(%{id: 1}))
      {:ok, position} = UmaDbClient.append(channel, [event])

      # Read events as a lazy stream
      {:ok, stream} = UmaDbClient.read(channel)
      Enum.each(stream, fn %UmaDb.V1.SequencedEvent{position: pos, event: e} ->
        IO.inspect({pos, e.event_type})
      end)

  ## TLS

  Pass gRPC credentials via `opts` to `connect/2`:

      cred = GRPC.Credential.new(ssl: [cacertfile: "/path/to/ca.pem"])
      {:ok, channel} = UmaDbClient.connect("myserver:443", cred: cred)

  ## Encoding and decoding payloads

  Event `data` is raw bytes, so callers normally encode and decode by hand. To
  configure that once, `use` this module to generate a client facade:

      defmodule MyApp.DB do
        use UmaDbClient, encode: &Jason.encode!/1, decode: &Jason.decode!/1
      end

      {:ok, channel} = MyApp.DB.connect("localhost:50051")

      event = MyApp.DB.event(type: "OrderPlaced", tags: ["order:1"], data: %{id: 1})
      {:ok, position} = MyApp.DB.append(channel, [event])

      {:ok, stream} = MyApp.DB.read(channel)
      Enum.each(stream, fn %UmaDbClient.Event{data: data} -> IO.inspect(data) end)

  See `__using__/1` for the full list of generated functions.
  """

  alias UmaDb.V1.DCB.Stub
  alias UmaDb.V1

  @type channel :: GRPC.Channel.t()
  @type position :: non_neg_integer()

  @doc """
  Generates a client facade with payload encoding and decoding configured.

  ## Options

  - `:encode` — applied to `event/1`'s `:data` before it is sent
  - `:decode` — applied to each event's `data` when read

  Both are optional and independent. Each accepts a one-argument function, a
  `{module, function}` tuple, or a module — exporting `encode!/1` for `:encode`
  and `decode!/1` for `:decode`, so `encode: Jason, decode: Jason` works.

  ## Generated functions

  - `event/1` — like `UmaDbClient.Builder.event/1`, but `:data` is a term run
    through `:encode`. Pass `:raw_data` instead to supply already-encoded bytes;
    passing both raises `ArgumentError`.
  - `read/2` and `subscribe/2` — return a lazy stream of `UmaDbClient.Event`
    structs with `data` decoded via `:decode`.
  - `connect/2`, `append/3`, `head/1`, `get_tracking_info/2` — delegate to this
    module.
  - `query/1`, `query_item/2`, `append_condition/1`, `tracking_info/2` —
    delegate to `UmaDbClient.Builder`.
  """
  defmacro __using__(opts) do
    encode = Keyword.get(opts, :encode)
    decode = Keyword.get(opts, :decode)

    quote do
      @doc """
      Builds an `UmaDb.V1.Event`, encoding `:data` with the configured encoder.
      """
      @spec event(keyword()) :: UmaDb.V1.Event.t()
      def event(opts) when is_list(opts) do
        opts
        |> UmaDbClient.Builder.__encode_opts__(unquote(encode))
        |> UmaDbClient.Builder.event()
      end

      @doc """
      Reads events, returning a lazy stream of decoded `UmaDbClient.Event` structs.
      """
      @spec read(UmaDbClient.channel(), keyword()) :: {:ok, Enumerable.t()} | {:error, term()}
      def read(channel, opts \\ []) do
        UmaDbClient.__decode_stream__(UmaDbClient.read(channel, opts), unquote(decode))
      end

      @doc """
      Subscribes to events, returning a lazy stream of decoded `UmaDbClient.Event` structs.
      """
      @spec subscribe(UmaDbClient.channel(), keyword()) ::
              {:ok, Enumerable.t()} | {:error, term()}
      def subscribe(channel, opts \\ []) do
        UmaDbClient.__decode_stream__(UmaDbClient.subscribe(channel, opts), unquote(decode))
      end

      defdelegate connect(target, opts \\ []), to: UmaDbClient
      defdelegate append(channel, events, opts \\ []), to: UmaDbClient
      defdelegate head(channel), to: UmaDbClient
      defdelegate get_tracking_info(channel, source), to: UmaDbClient

      defdelegate query(items), to: UmaDbClient.Builder
      defdelegate query_item(types \\ [], tags \\ []), to: UmaDbClient.Builder
      defdelegate append_condition(opts \\ []), to: UmaDbClient.Builder
      defdelegate tracking_info(source, position), to: UmaDbClient.Builder
    end
  end

  @doc false
  def __decode_stream__({:ok, stream}, decode) do
    {:ok, Stream.map(stream, &UmaDbClient.Event.from_sequenced(&1, decode))}
  end

  def __decode_stream__({:error, _} = err, _decode), do: err

  @doc """
  Opens a gRPC channel to a UmaDB server.

  `target` is a `"host:port"` string. `opts` are passed directly to
  `GRPC.Stub.connect/2` (e.g. `cred:` for TLS, `interceptors:` etc.).
  """
  @spec connect(String.t(), keyword()) :: {:ok, channel()} | {:error, term()}
  def connect(target, opts \\ []) do
    GRPC.Stub.connect(target, opts)
  end

  @doc """
  Returns the current head position of the event log.

  Returns `{:ok, nil}` when the log is empty.
  """
  @spec head(channel()) :: {:ok, position() | nil} | {:error, term()}
  def head(channel) do
    case Stub.head(channel, %V1.HeadRequest{}) do
      {:ok, %V1.HeadResponse{position: pos}} -> {:ok, pos}
      {:error, _} = err -> err
    end
  end

  @doc """
  Returns the last tracked position for the given source identifier.

  Returns `{:ok, nil}` when no tracking record exists for `source`.
  """
  @spec get_tracking_info(channel(), String.t()) :: {:ok, position() | nil} | {:error, term()}
  def get_tracking_info(channel, source) do
    case Stub.get_tracking_info(channel, %V1.TrackingRequest{source: source}) do
      {:ok, %V1.TrackingResponse{position: pos}} -> {:ok, pos}
      {:error, _} = err -> err
    end
  end

  @doc """
  Appends one or more events to the log.

  Returns `{:ok, position}` where `position` is the log position after the append.

  ## Options

  - `:condition` — an `UmaDb.V1.AppendCondition` struct for optimistic concurrency
  - `:tracking_info` — an `UmaDb.V1.TrackingInfo` struct to update a tracking cursor
  """
  @spec append(channel(), [V1.Event.t()], keyword()) :: {:ok, position()} | {:error, term()}
  def append(channel, events, opts \\ []) do
    request = %V1.AppendRequest{
      events: events,
      condition: Keyword.get(opts, :condition),
      tracking_info: Keyword.get(opts, :tracking_info)
    }

    case Stub.append(channel, request) do
      {:ok, %V1.AppendResponse{position: pos}} -> {:ok, pos}
      {:error, _} = err -> err
    end
  end

  @doc """
  Returns a lazy `Enumerable` of `UmaDb.V1.SequencedEvent` structs from the log.

  The stream is server-driven: each element is received from the server as it is
  produced. Iterate with `Enum.to_list/1`, `Stream.each/2`, etc.

  Mid-stream gRPC errors raise `GRPC.RPCError`. Wrap iteration in `try/rescue`
  if you need to recover from them.

  ## Options

  - `:query` — `UmaDb.V1.Query` to filter by event type and/or tags
  - `:start` — starting position (inclusive)
  - `:backwards` — if `true`, read in reverse order
  - `:limit` — maximum number of events to return
  - `:subscribe` — if `true`, keep the stream open for new events after catching up
  - `:batch_size` — number of events per server response batch
  """
  @spec read(channel(), keyword()) :: {:ok, Enumerable.t()} | {:error, term()}
  def read(channel, opts \\ []) do
    request = %V1.ReadRequest{
      query: Keyword.get(opts, :query),
      start: Keyword.get(opts, :start),
      backwards: Keyword.get(opts, :backwards),
      limit: Keyword.get(opts, :limit),
      subscribe: Keyword.get(opts, :subscribe),
      batch_size: Keyword.get(opts, :batch_size)
    }

    case Stub.read(channel, request) do
      {:ok, grpc_stream} ->
        {:ok, flatten_sequenced_events(grpc_stream, V1.ReadResponse)}

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

  @doc """
  Returns a lazy `Enumerable` of `UmaDb.V1.SequencedEvent` structs, starting
  from the given position and continuing indefinitely as new events arrive.

  This is a long-running server-push stream. Blocking iteration (e.g. via
  `Enum.each/2`) will not return until the server closes the stream or an
  error occurs. Consider running it in a dedicated process.

  ## Options

  - `:query` — `UmaDb.V1.Query` to filter by event type and/or tags
  - `:after` — only receive events after this position
  - `:batch_size` — number of events per server response batch
  """
  @spec subscribe(channel(), keyword()) :: {:ok, Enumerable.t()} | {:error, term()}
  def subscribe(channel, opts \\ []) do
    request = %V1.SubscribeRequest{
      query: Keyword.get(opts, :query),
      after: Keyword.get(opts, :after),
      batch_size: Keyword.get(opts, :batch_size)
    }

    case Stub.subscribe(channel, request) do
      {:ok, grpc_stream} ->
        {:ok, flatten_sequenced_events(grpc_stream, V1.SubscribeResponse)}

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

  defp flatten_sequenced_events(grpc_stream, response_module) do
    Stream.flat_map(grpc_stream, fn
      {:ok, response} ->
        Map.fetch!(response, :events)

      {:error, reason} ->
        raise GRPC.RPCError,
          status: GRPC.Status.internal(),
          message: "stream error from #{inspect(response_module)}: #{inspect(reason)}"
    end)
  end
end