<commit_before>defmodule Service.Domo do
  @behaviour Reducer

  import DB.Validation, only: [flip_v1_uuid: 1]

  alias DB.Event
  alias Reducer.State
  require Logger


  @domains [:domo]
  @types [:pull]
  def domains, do: @domains
  def types, do: @types

  def correct_type?(event) do
    Enum.member?([
      "pull"], event.type)
  end


  @spec call(list(Event.t), State.t) :: State.t
  def call(events, state)do
    {model, new_events} = Enum.filter(events, fn(event) -> correct_type?(event) end)
     |> domo_service_recursive({state, []})
    %State{model: model, new_events: new_events}
  end


  def domo_service_recursive([], {model, new_events}), do: {model, new_events}
  def domo_service_recursive([event | remaining_events], {model, new_events}) do
    domo_service_recursive(remaining_events, domo_service(event, {model, new_events}))
  end


  def domo_service(event, {state, new_events}) do
    store_ids = get_store_ids()
    model = state.model
    meta = event.meta
    dataset_id = meta["dataset_id"]
    client_id = meta["client_id"]
    client_secret = meta["client_secret"]
    type = meta["out_going_type"]
    last_played = flip_v1_uuid(event.event_id)

    domo_dataset(dataset_id, client_id, client_secret)
      |> hash_file(model, type, store_ids)
      |> return_model_and_events(last_played)


  end

  def domo_dataset(dataset_id, client_id, client_secret) do
   get_access_token(client_id, client_secret)
    |> get_dataset(dataset_id)
  end

  def get_access_token(client_id, client_secret) do
    url = "https://api.domo.com/oauth/token?grant_type=client_credentials&scope=data"

    case HTTPoison.get(url, %{}, [hackney: [basic_auth: {client_id, client_secret}]]) do
      {:ok, %HTTPoison.Response{status_code: 200, body: body}} ->
        response = Poison.Parser.parse!(body)
        response["access_token"]
      {:ok, %HTTPoison.Response{status_code: code}} ->
        Logger.error("token error, status_code#{code}")
      {:error, %HTTPoison.Error{reason: reason}} ->
        Logger.error("token error, #{reason}")
    end

  end

  def get_dataset(access_token, dataset_id) do
    url = "https://api.domo.com/v1/datasets/#{dataset_id}/data?includeHeader=true&fileName=data_dump.csv"

    case HTTPoison.get(url, [{"Authorization", "bearer #{access_token}"}]) do
      {:ok, %HTTPoison.Response{status_code: 200, body: body}} ->
        body
        {:ok, %HTTPoison.Response{status_code: code}} ->
          Logger.error("dataset error, status_code#{code}")
        {:error, %HTTPoison.Error{reason: reason}} ->
          Logger.error("dataset error, #{reason}")
    end

  end

  def hash_file(body, model, type, store_ids) do
    hash_state = is_empty?(model)
    [col_heads | values] = String.split(body, "\n", parts: 2)
    file = String.split(to_string(values), "\n")
    file |> Enum.reduce({[], col_heads, type, hash_state, store_ids}, &reducer/2)
  end

  def reducer(row, {events, col_heads, type, hash_state, store_ids}) when row != "" do
    IO.inspect(row)
    IO.inspect(hash_state)
    {new_events, col_heads, type, new_hash_state, store_ids} =
      case Donethat.new?(row, hash_state) do
        {true, new_hash_state} ->
          {[make_event(col_heads, type, row, store_ids) | events], col_heads, type, new_hash_state, store_ids}
        {_, new_hash_state} ->
          {events, col_heads, type, new_hash_state, store_ids}
      end
    {new_events, col_heads, type, new_hash_state, store_ids}
  end

  def reducer(row, {events, col_heads, type, hash_state, store_ids}) do
    {events, col_heads, type, hash_state, store_ids}
  end

  def is_empty?(model) when map_size(model) == 0 do
    Donethat.empty_state
  end

  def is_empty?(model) do
    model[:hash_state]
  end

  def make_event(col_heads, type, row, store_ids) do
    [store | _t] = String.split(row, ",", parts: 2)
    meta = build_meta_map(col_heads, row)
    event = %Event{domain: "stats",<commit_msg>domo service should be error free<commit_after>defmodule Service.Domo do
  @behaviour Reducer

  import DB.Validation, only: [flip_v1_uuid: 1]

  alias DB.Event
  alias Reducer.State
  require Logger


  @domains [:domo]
  @types [:pull]
  def domains, do: @domains
  def types, do: @types

  def correct_type?(event) do
    Enum.member?([
      "pull"], event.type)
  end


  @spec call(list(Event.t), State.t) :: State.t
  def call(events, state)do
    {model, new_events} = Enum.filter(events, fn(event) -> correct_type?(event) end)
     |> domo_service_recursive({state, []})
    %State{model: model, new_events: new_events}
  end


  def domo_service_recursive([], {model, new_events}), do: {model, new_events}
  def domo_service_recursive([event | remaining_events], {model, new_events}) do
    domo_service_recursive(remaining_events, domo_service(event, {model, new_events}))
  end


  def domo_service(event, {state, new_events}) do
    store_ids = get_store_ids()
    model = state.model
    meta = event.meta
    dataset_id = meta["dataset_id"]
    client_id = meta["client_id"]
    client_secret = meta["client_secret"]
    type = meta["out_going_type"]
    last_played = flip_v1_uuid(event.event_id)

    domo_dataset(dataset_id, client_id, client_secret)
      |> hash_file(model, type, store_ids)
      |> return_model_and_events(last_played)


  end

  def domo_dataset(dataset_id, client_id, client_secret) do
   get_access_token(client_id, client_secret)
    |> get_dataset(dataset_id)
  end

  def get_access_token(client_id, client_secret) do
    url = "https://api.domo.com/oauth/token?grant_type=client_credentials&scope=data"

    case HTTPoison.get(url, %{}, [hackney: [basic_auth: {client_id, client_secret}]]) do
      {:ok, %HTTPoison.Response{status_code: 200, body: body}} ->
        response = Poison.Parser.parse!(body)
        response["access_token"]
      {:ok, %HTTPoison.Response{status_code: code}} ->
        Logger.error("token error, status_code#{code}")
      {:error, %HTTPoison.Error{reason: reason}} ->
        Logger.error("token error, #{reason}")
    end

  end

  def get_dataset(access_token, dataset_id) do
    url = "https://api.domo.com/v1/datasets/#{dataset_id}/data?includeHeader=true&fileName=data_dump.csv"

    case HTTPoison.get(url, [{"Authorization", "bearer #{access_token}"}]) do
      {:ok, %HTTPoison.Response{status_code: 200, body: body}} ->
        body
        {:ok, %HTTPoison.Response{status_code: code}} ->
          Logger.error("dataset error, status_code#{code}")
        {:error, %HTTPoison.Error{reason: reason}} ->
          Logger.error("dataset error, #{reason}")
    end

  end

  def hash_file(body, model, type, store_ids) do
    hash_state = is_empty?(model)
    [col_heads | values] = String.split(body, "\n", parts: 2)
    file = String.split(to_string(values), "\n")
    file |> Enum.reduce({[], col_heads, type, hash_state, store_ids}, &reducer/2)
  end

  def reducer(row, {events, col_heads, type, hash_state, store_ids}) when row != "" do
    {new_events, col_heads, type, new_hash_state, store_ids} =
      case Donethat.new?(row, hash_state) do
        {true, new_hash_state} ->
          {[make_event(col_heads, type, row, store_ids) | events], col_heads, type, new_hash_state, store_ids}
        {_, new_hash_state} ->
          {events, col_heads, type, new_hash_state, store_ids}
      end
    {new_events, col_heads, type, new_hash_state, store_ids}
  end

  def reducer(row, {events, col_heads, type, hash_state, store_ids}) do
    {events, col_heads, type, hash_state, store_ids}
  end

  def is_empty?(model) when map_size(model) == 0 do
    Donethat.empty_state
  end

  def is_empty?(model) do
    to_struct(HashState, model["hash_state"])
  end

  def to_struct(kind, attrs) do
  struct = struct(kind)
  Enum.reduce Map.to_list(struct), struct, fn {k, _}, acc ->
    case Map.fetch(attrs, Atom.to_string(k)) do
      {:ok, v} -> %{acc | k => v}
      :error -> acc
    end
  end
end

  def make_event(col_heads, type, row, store_ids) do
    [store | _t] = String.split(row, ",", parts: 2)
    meta = build_meta_map(col_heads, row)
    event = %Event{domain: "stats",
{'fraction_non_alphanumeric': 0.1377097729516288, 'fraction_numerical': 0.0034550839091806516, 'mean_word_length': 3.8359188544152745, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 35, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3207679', 'n_tokens_mistral': 2934, 'n_tokens_neox': 2868, 'n_words': 611}