kane

defmodule Kane.Message do
  alias Kane.Topic
  alias Kane.Client.Response.Error

  @type t :: %__MODULE__{
    id: String.t | nil,
    attributes: Map.t,
    data: any,
    ack_id: String.t | nil,
    publish_time: String.t | nil
  }

  defstruct id: nil, attributes: %{}, data: nil, ack_id: nil, publish_time: nil

  @spec publish(binary, binary) :: {:ok, t} | Error.t
  def publish(message, topic) when is_binary(message) and is_binary(topic) do
    publish(%__MODULE__{data: message}, %Topic{name: topic})
  end

  @spec publish(t, Topic.t) :: {:ok, t} | Error.t
  def publish(%__MODULE__{}=message, %Topic{}=topic) do
    case publish([message], topic) do
      {:ok, [message|_]} -> {:ok, message}
      err -> err
    end
  end

  @spec publish([t], Topic.t) :: {:ok, [t]} | Error.t
  def publish(messages, %Topic{name: topic}) when is_list(messages)  do
    case Kane.Client.post(path(topic), data(messages)) do
      {:ok, body, _code} ->
        collected = body
        |> Poison.decode!
        |> Map.get("messageIds")
        |> Stream.with_index
        |> Enum.map(fn({id, i}) ->
             %{Enum.at(messages, i) | id: id}
           end)

        {:ok, collected}
      err -> err
    end
  end

  @spec data(t) :: map
  def data(%__MODULE__{}=message), do: data([message])

  @spec data([t]) :: map
  def data(messages) when is_list(messages) do
    %{
      "messages" =>
        Enum.map(messages, fn(%__MODULE__{data: d, attributes: a}) ->
          %{
            "data" => encode_body(d),
            "attributes" => Enum.reduce(a, %{}, fn({key, val}, map)->
              Map.put(map, key, val)
            end)
          }
        end)
    }
  end

  @spec encode_body(any) :: binary
  def encode_body(body) when is_binary(body), do: Base.encode64(body)
  def encode_body(body), do: body |> Poison.encode! |> encode_body

  def json(%__MODULE__{}=message), do: json([message])
  def json(messages) when is_list(messages) do
    data(messages) |> Poison.encode!
  end

  def from_subscription!(%{}=data) do
    {:ok, message} = from_subscription(data)
    message
  end

  def from_subscription(%{"ackId" => ack, "message" => %{"data" => data, "publishTime" => time, "messageId" => id} = message }) do
    attr = Map.get(message, "attributes", %{})
    {:ok, %__MODULE__{
      id: id,
      publish_time: time,
      ack_id: ack,
      data: data |> Base.decode64!,
      attributes: attr
    }}
  end

  defp path(%Topic{name: topic}), do: path(topic)
  defp path(topic) do
    {:ok, project} = Goth.Config.get(:project_id)
    "projects/#{project}/topics/#{Topic.strip!(topic)}:publish"
  end
end