smart_city

defmodule SmartCity.Data do
  @moduledoc """
  Message struct shared amongst all SmartCity microservices.

  ```javascript
  const DataMessage = {
      "dataset_id": "",        // UUID
      "payload": {},
      "_metadata": {           // cannot be used safely
          "orgName": "",       // ~r/^[a-zA-Z_]+$/
          "dataName": "",      // ~r/^[a-zA-Z_]+$/
          "stream": true
      },
      "operational": {
          "timing": [{
              "startTime": "", // iso8601
              "endTime": "",   // iso8601
              "app": "",       // microservice generating timing data
              "label": ""      // label for this particular timing data
          }]
      }
  }
  ```
  """

  alias SmartCity.Data
  alias SmartCity.Data.Timing
  alias SmartCity.Helpers

  @type t :: %SmartCity.Data{
          :dataset_id => String.t(),
          :operational => %{
            :timing => list(SmartCity.Data.Timing.t())
          },
          :payload => String.t(),
          :_metadata => %{
            :org => String.t(),
            :name => String.t(),
            :stream => boolean()
          },
          :version => String.t()
        }
  @type payload :: String.t()

  @derive Jason.Encoder
  @enforce_keys [:dataset_id, :payload, :_metadata, :operational]
  defstruct version: "0.1",
            _metadata: %{org: nil, name: nil, stream: false},
            dataset_id: nil,
            payload: nil,
            operational: %{timing: []}

  @doc """
  Returns a new `SmartCity.Data` struct. `SmartCity.Data.Timing`
    structs will be created along the way.

  Can be created from:
    - map with string keys
    - map with atom keys
    - JSON

  ## Examples

      iex> SmartCity.Data.new(%{dataset_id: "a_guid", payload: "the_data", _metadata: %{org: "scos", name: "example"}, operational: %{timing: [%{app: "app name", label: "function name", start_time: "2019-05-06T19:51:41+00:00", end_time: "2019-05-06T19:51:51+00:00"}]}})
      {:ok, %SmartCity.Data{
          dataset_id: "a_guid",
          payload: "the_data",
          _metadata: %{org: "scos", name: "example"},
          operational: %{
          timing: [%SmartCity.Data.Timing{ app: "app name", end_time: "2019-05-06T19:51:51+00:00", label: "function name", start_time: "2019-05-06T19:51:41+00:00"}]
        }
      }}
  """
  @spec new(map() | String.t()) :: {:ok, SmartCity.Data.t()} | {:error, String.t()}
  def new(msg) when is_binary(msg) do
    with {:ok, decoded} <- Jason.decode(msg) do
      new(decoded)
    end
  end

  def new(%{"dataset_id" => _} = msg) do
    %{
      dataset_id: msg["dataset_id"],
      operational: Helpers.to_atom_keys(msg["operational"]),
      payload: msg["payload"],
      _metadata: Helpers.to_atom_keys(msg["_metadata"])
    }
    |> new()
  end

  def new(%{
        dataset_id: dataset_id,
        operational: operational,
        payload: payload,
        _metadata: metadata
      }) do
    timings = Map.get(operational, :timing, [])

    struct =
      struct(__MODULE__, %{
        dataset_id: dataset_id,
        payload: payload,
        _metadata: metadata,
        operational: %{operational | timing: Enum.map(timings, &Timing.new/1)}
      })

    {:ok, struct}
  rescue
    e -> {:error, e}
  end

  def new(msg) do
    {:error, "Invalid data message: #{inspect(msg)}"}
  end

  @doc """
  Defines the string that will be the payload of the last message in a dataset.
  """
  defmacro end_of_data(), do: quote(do: "END_OF_DATA")

  @doc """
  Encodes `SmartCity.Data` into JSON. Typically used right before sending as a Kafka message.
  """
  @spec encode(SmartCity.Data.t()) ::
          {:ok, String.t()} | {:error, Jason.EncodeError.t() | Exception.t()}
  def encode(%__MODULE__{} = message) do
    Jason.encode(message)
  end

  @doc """
  Encodes `SmartCity.Data` into JSON. Typically used right before sending as a Kafka message.

  Raises an error if it fails to convert to a JSON string.
  """
  @spec encode!(SmartCity.Data.t()) :: String.t()
  def encode!(%__MODULE__{} = message) do
    Jason.encode!(message)
  end

  @doc """
  Adds a `SmartCity.Data.Timing` to the list of timings in this `SmartCity.Data`. The timing will be validated to ensure both start and end times have been set.

  Returns a `SmartCity.Data` struct with `new_timing` prepended to existing timings list.

  ## Parameters
    - message: A `SmartCity.Data`
    - new_timing: A timing you want to add. Must have `start_time` and `end_time` set
  """
  @spec add_timing(
          SmartCity.Data.t(),
          SmartCity.Data.Timing.t()
        ) :: SmartCity.Data.t()
  def add_timing(
        %__MODULE__{operational: %{timing: timing}} = message,
        %Data.Timing{} = new_timing
      ) do
    case Timing.validate(new_timing) do
      {:ok, new_timing} -> put_in_operational(message, :timing, [new_timing | timing])
      {:error, errors} -> raise ArgumentError, "Invalid Timing: #{errors}"
    end
  end

  @doc """
  Creates a new `SmartCity.Data` struct using `new/1` and adds timing information to the message.

  Returns a `SmartCity.Data` struct with `new_timing` prepended to existing timings list.

  ## Parameters
    - message: A `SmartCity.Data`
    - app: The application that is asking to create the new `SmartCity.Data`. Ex. `reaper` or `voltron`
  """
  @spec timed_new(map(), String.t()) :: {:ok, SmartCity.Data.t()} | {:error, String.t()}
  def timed_new(msg, app) do
    label = inspect(&Data.new/1)

    case Timing.measure(app, label, fn -> new(msg) end) do
      {:ok, msg, timing} -> {:ok, msg |> add_timing(timing)}
      error -> error
    end
  end

  @doc """
  Transforms the `SmartCity.Data` `payload` field with the given unary function and replaces it in the message.

  Additionally, returns a `SmartCity.Data` struct with `new_timing` prepended to existing timings list.

  ## Parameters
    - message: A `SmartCity.Data`
    - app: The application that is asking to create the new `SmartCity.Data`. Ex. `reaper` or `voltron`
    - function: an arity 1 (/1) function that will transform the payload in the provided message
  """
  @spec timed_transform(
          SmartCity.Data.t(),
          String.t(),
          (payload() -> {:ok, term()} | {:error, term()})
        ) :: {:ok, SmartCity.Data.t()} | {:error, String.t()}
  def timed_transform(%Data{} = msg, app, function) when is_function(function, 1) do
    label = inspect(function)

    case Timing.measure(app, label, fn -> function.(msg.payload) end) do
      {:ok, result, timing} -> {:ok, msg |> add_timing(timing) |> Map.replace!(:payload, result)}
      error -> error
    end
  end

  @doc """
  Get all timings on this Data

  Returns a list of  `SmartCity.Data.Timing` structs or `[]`

  ## Parameters
    - data_message: The message to extract timings from
  """
  @spec get_all_timings(SmartCity.Data.t()) :: list(SmartCity.Data.Timing.t())
  def get_all_timings(%__MODULE__{operational: %{timing: timing}}), do: timing

  # Private functions
  defp put_in_operational(%__MODULE__{operational: operational} = message, key, value) do
    %{message | operational: Map.put(operational, key, value)}
  end
end