eventstore

defmodule EventStore.Storage.Stream do
  @moduledoc false

  alias EventStore.Page
  alias EventStore.Sql.Statements
  alias EventStore.Streams.StreamInfo

  def stream_info(conn, stream_uuid, opts) do
    {schema, opts} = Keyword.pop(opts, :schema)

    query = Statements.query_stream_info(schema)

    case Postgrex.query(conn, query, [stream_uuid], opts) do
      {:ok, %Postgrex.Result{num_rows: 0}} ->
        {:ok, StreamInfo.new(stream_uuid)}

      {:ok, %Postgrex.Result{rows: [row]}} ->
        {:ok, to_stream_info(row)}

      {:error, _error} = reply ->
        reply
    end
  end

  def paginate_streams(conn, opts) do
    {page_number, opts} = Keyword.pop(opts, :page_number, 1)
    {page_size, opts} = Keyword.pop(opts, :page_size, 50)
    {schema, opts} = Keyword.pop(opts, :schema)

    offset = page_size * (page_number - 1)

    with {:ok, total_streams} <- count_streams(conn, schema, opts),
         {:ok, streams} <- query_streams(conn, schema, page_size, offset, opts) do
      page = %Page{
        entries: streams,
        page_number: page_number,
        page_size: page_size,
        total_entries: total_streams,
        total_pages: Page.total_pages(total_streams, page_size)
      }

      {:ok, page}
    end
  end

  defp count_streams(conn, schema, opts) do
    query = Statements.count_streams(schema)

    search = search_term(opts)

    case Postgrex.query(conn, query, [search], opts) do
      {:ok, %Postgrex.Result{rows: [[count]]}} ->
        {:ok, count}

      {:error, _error} = reply ->
        reply
    end
  end

  defp query_streams(conn, schema, limit, offset, opts) do
    search = search_term(opts)

    sort_by =
      case Keyword.get(opts, :sort_by, :stream_id) do
        :stream_uuid -> "stream_uuid"
        :stream_id -> "stream_id"
        :stream_version -> "stream_version"
        :created_at -> "created_at"
        :deleted_at -> "deleted_at"
        :status -> "status"
      end

    sort_dir =
      case Keyword.get(opts, :sort_dir, :asc) do
        :asc -> "ASC"
        :desc -> "DESC"
      end

    query = Statements.query_streams(schema, sort_by, sort_dir)

    case Postgrex.query(conn, query, [search, limit, offset], opts) do
      {:ok, %Postgrex.Result{num_rows: 0}} ->
        {:ok, []}

      {:ok, %Postgrex.Result{rows: rows}} ->
        {:ok, Enum.map(rows, &to_stream_info/1)}

      {:error, _error} = reply ->
        reply
    end
  end

  defp search_term(opts) do
    case Keyword.get(opts, :search) do
      nil -> nil
      "" -> nil
      search when is_binary(search) -> "%" <> search <> "%"
    end
  end

  defp to_stream_info(row) do
    [stream_id, stream_uuid, stream_version, created_at, deleted_at] = row

    stream_version = if is_nil(stream_version), do: 0, else: stream_version
    status = if is_nil(deleted_at), do: :created, else: :deleted

    %StreamInfo{
      stream_uuid: stream_uuid,
      stream_id: stream_id,
      stream_version: stream_version,
      created_at: created_at,
      deleted_at: deleted_at,
      status: status
    }
  end
end