instream_with_delete

defmodule Instream.Series.Hydrator do
  @moduledoc false

  alias Instream.Decoder.RFC3339

  @doc """
  Converts a plain map into a series definition struct.

  Keys not defined in the series are silently dropped.
  """
  @spec from_map(module, map) :: struct
  def from_map(series, data) do
    data_fields = Map.take(data, series.__meta__(:fields))
    data_tags = Map.take(data, series.__meta__(:tags))

    struct(series, %{
      fields: struct(Module.safe_concat(series, Fields), data_fields),
      tags: struct(Module.safe_concat(series, Tags), data_tags),
      timestamp: convert_to_timestamp(data[:time] || data[:timestamp])
    })
  end

  @doc """
  Converts a query result map into a list of series definition structs.

  Keys not defined in the series are silently dropped.
  """
  @spec from_result(module, map | [map]) :: [struct]
  def from_result(series, %{
        results: [%{series: [%{values: result_values, columns: columns} = data]}]
      }) do
    # optional :tags set in InfluxQL "GROUP BY" results
    tags = Map.get(data, :tags, %{})

    Enum.map(result_values, fn values ->
      mapped_values =
        columns
        |> Enum.zip(values)
        |> Enum.into(%{}, fn {k, v} -> {String.to_atom(k), v} end)

      from_map(series, Map.merge(tags, mapped_values))
    end)
  end

  def from_result(series, rows) when is_list(rows) do
    Enum.map(rows, fn row ->
      row =
        case row["_field"] do
          nil -> row
          field -> Map.put(row, field, row["_value"])
        end

      timestamp = row["_time"]

      data =
        row
        |> Map.drop(["_field", "_measurement", "_start", "_stop", "_time", "_value", "table"])
        |> Map.new(fn {k, v} -> {String.to_atom(k), v} end)
        |> Map.put(:time, timestamp)

      from_map(series, data)
    end)
  end

  defp convert_to_timestamp(time) when is_integer(time), do: time
  defp convert_to_timestamp(time) when is_binary(time), do: RFC3339.to_nanosecond(time)
  defp convert_to_timestamp(_), do: nil
end