Create player table, track player updates, fix match processing tracking

apps/lol_analytics/lib/lol_analytics/match/match_repo.ex

@@ -3,6 +3,11 @@ defmodule LolAnalytics.Match.MatchRepo do
 
   import Ecto.Query
 
+  def list_matches do
+    query = from m in MatchSchema, order_by: [desc: m.match_id]
+    LoLAnalytics.Repo.all(query)
+  end
+
   @spec get_match(String.t()) :: %LolAnalytics.Match.MatchSchema{}
   def get_match(match_id) do
     query = from m in MatchSchema, where: m.match_id == ^match_id

apps/lol_analytics/lib/lol_analytics/player/player_repo.ex

@@ -0,0 +1,27 @@
+defmodule LolAnalytics.Player.PlayerRepo do
+  alias LolAnalytics.Player.PlayerSchema
+
+  import Ecto.Query
+
+  def list_players do
+    query = from(p in PlayerSchema)
+    LoLAnalytics.Repo.all(query)
+  end
+
+  def get_player(puuid) do
+    query = from p in PlayerSchema, where: p.puuid == ^puuid
+    LoLAnalytics.Repo.one(query)
+  end
+
+  def insert_player(puuid) do
+    %PlayerSchema{}
+    |> PlayerSchema.changeset(%{puuid: puuid, region: "EUW"})
+    |> LoLAnalytics.Repo.insert()
+  end
+
+  def update_player(player, attrs) do
+    player
+    |> PlayerSchema.changeset(attrs)
+    |> LoLAnalytics.Repo.update()
+  end
+end

apps/lol_analytics/lib/lol_analytics/player/player_schema.ex

@@ -0,0 +1,22 @@
+defmodule LolAnalytics.Player.PlayerSchema do
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  schema "player" do
+    field :puuid, :string
+    field :region, :string
+
+    field :last_processed_at, :utc_datetime,
+      default: DateTime.utc_now() |> DateTime.truncate(:second)
+
+    timestamps()
+  end
+
+  @spec changeset(%__MODULE__{}) ::
+          Ecto.Changeset.t()
+  def changeset(player = %__MODULE__{}, params \\ %{}) do
+    player
+    |> cast(params, [:puuid, :region, :last_processed_at])
+    |> validate_required([:puuid, :region, :last_processed_at])
+  end
+end

apps/lol_analytics/priv/repo/migrations/20240502203545_player.exs

@@ -0,0 +1,14 @@
+defmodule LoLAnalytics.Repo.Migrations.Player do
+  use Ecto.Migration
+
+  def change do
+    create table("player") do
+      add :puuid, :string
+      add :region, :string
+      add :last_processed_at, :utc_datetime
+      timestamps()
+    end
+
+    create index(:player, [:puuid])
+  end
+end

apps/scrapper/lib/scrapper/api/match_api.ex

@@ -11,6 +11,7 @@ defmodule Scrapper.Data.Api.MatchApi do
   @spec get_match_by_id(String.t()) :: %Scrapper.Api.Model.MatchResponse{}
   def get_match_by_id(match_id) do
     url = String.replace(@match_base_endpoint, "%{matchid}", match_id)
+    Logger.info("Making request to #{url}")
     api_key = System.get_env("RIOT_API_KEY")
     headers = [{"X-Riot-Token", api_key}]
     response = HTTPoison.get!(url, headers, timeout: 5000)
@@ -25,9 +26,15 @@ defmodule Scrapper.Data.Api.MatchApi do
     end
   end
 
+  @doc """
+  Get matches from player
+
+  iex> Scrapper.Data.Api.MatchApi.get_matches_from_player "JB6TdEWlKjZwnbgdSzOogYepNfjLPdUh68S8b4kUu4EEZy4R4MMAgv92QMj1XgVjtzHmZVLaOW7mzg"
+  """
   @spec get_matches_from_player(String.t()) :: list(String.t()) | integer()
   def get_matches_from_player(puuid) do
-    url = String.replace(@puuid_matches_base_endpoint, "%{puuid}", puuid)
+    url = String.replace(@puuid_matches_base_endpoint, "%{puuid}", URI.encode(puuid))
+    Logger.info("Making request to #{url}")
     api_key = System.get_env("RIOT_API_KEY")
     headers = [{"X-Riot-Token", api_key}]
     response = HTTPoison.get!(url, headers, timeout: 5000)
@@ -37,6 +44,7 @@ defmodule Scrapper.Data.Api.MatchApi do
         {:ok, Poison.decode!(response.body)}
 
       code ->
+        IO.inspect(response)
         Logger.error("Error getting matches from player #{puuid} #{code}")
         {:err, response.status_code}
     end

apps/scrapper/lib/scrapper/processor/match_processor.ex

@@ -16,17 +16,17 @@ defmodule Scrapper.Processor.MatchProcessor do
            ],
            on_failure: :reject,
            qos: [
-             prefetch_count: 3
+             prefetch_count: 1
            ]},
         concurrency: 1,
         rate_limiting: [
-          interval: 1000 * 60,
+          interval: 1000 * 3,
-          allowed_messages: 150
+          allowed_messages: 1
         ]
       ],
       processors: [
         default: [
-          concurrency: 5
+          concurrency: 1
         ]
       ]
     )
@@ -45,19 +45,19 @@ defmodule Scrapper.Processor.MatchProcessor do
   def process_resp({:ok, raw_match}, match_id) do
     decoded_match = Poison.decode!(raw_match, as: %Scrapper.Api.Model.MatchResponse{})
     Scrapper.Storage.S3MatchStorage.store_match(match_id, raw_match)
+    match = LolAnalytics.Match.MatchRepo.get_match(match_id)
+
+    case match do
+      nil ->
+        LolAnalytics.Match.MatchRepo.insert_match(match_id)
+
+      _ ->
+        LolAnalytics.Match.MatchRepo.update_match(match, %{:processed => true})
+    end
 
     decoded_match.metadata.participants
     |> Enum.each(fn participant ->
-      case Scrapper.Data.Api.MatchApi.get_matches_from_player(participant) do
+      Scrapper.Queue.PlayerQueue.queue_player(participant)
-        {:ok, matches} ->
-          matches
-          |> Enum.each(fn match_id ->
-            Scrapper.Queue.MatchQueue.queue_match(match_id)
-          end)
-
-        {:err, code} ->
-          {:err, code}
-      end
     end)
   end
 

apps/scrapper/lib/scrapper/processor/player_processor.ex

@@ -1,4 +1,5 @@
 defmodule Scrapper.Processor.PlayerProcessor do
+  alias Calendar.ISO
   use Broadway
 
   def start_link(_opts) do
@@ -16,17 +17,17 @@ defmodule Scrapper.Processor.PlayerProcessor do
            ],
            on_failure: :reject,
            qos: [
-             prefetch_count: 3
+             prefetch_count: 1
            ]},
         concurrency: 1,
         rate_limiting: [
-          interval: 1000 * 60,
+          interval: 1000 * 10,
-          allowed_messages: 5
+          allowed_messages: 1
         ]
       ],
       processors: [
         default: [
-          concurrency: 5
+          concurrency: 1
         ]
       ]
     )
@@ -38,6 +39,17 @@ defmodule Scrapper.Processor.PlayerProcessor do
 
     resp = Scrapper.Data.Api.MatchApi.get_matches_from_player(puuid)
 
+    case LolAnalytics.Player.PlayerRepo.get_player(puuid) do
+      nil ->
+        LolAnalytics.Player.PlayerRepo.insert_player(puuid)
+
+      player ->
+        player
+        |> LolAnalytics.Player.PlayerRepo.update_player(%{
+          :last_processed_at => DateTime.utc_now() |> DateTime.truncate(:seconds)
+        })
+    end
+
     case resp do
       {:ok, matches} ->
         {
@@ -53,4 +65,12 @@ defmodule Scrapper.Processor.PlayerProcessor do
 
     message
   end
+
+  defp update_player(nil), do: :player_not_found
+
+  defp update_player(player) do
+    LolAnalytics.Player.PlayerRepo.update_player(player, %{
+      :last_processed_at => DateTime.utc_now() |> DateTime.truncate(:second)
+    })
+  end
 end

apps/scrapper/lib/scrapper/queue/match_queue.ex

@@ -15,7 +15,11 @@ defmodule Scrapper.Queue.MatchQueue do
 
   @spec queue_match(String.t()) :: any()
   def queue_match(match_id) do
-    GenServer.call(__MODULE__, {:queue_match, match_id})
+    LolAnalytics.Match.MatchRepo.get_match(match_id)
+    |> case do
+      nil -> GenServer.call(__MODULE__, {:queue_match, match_id})
+      match -> :already_processed
+    end
   end
 
   def handle_call({:queue_match, match_id}, _from, %{:channel => channel} = state) do

apps/scrapper/lib/scrapper/queue/player_queue.ex

@@ -14,7 +14,11 @@ defmodule Scrapper.Queue.PlayerQueue do
 
   @spec queue_player(String.t()) :: nil
   def queue_player(puuid) do
-    GenServer.call(__MODULE__, {:queue_player, puuid})
+    case LolAnalytics.Player.PlayerRepo.get_player(puuid) do
+      nil -> GenServer.call(__MODULE__, {:queue_player, puuid})
+      _ -> :already_processed
+    end
+
   end
 
   def handle_call({:queue_player, puuid}, _from, {channel, _} = state) do

apps/scrapper/mix.exs

@@ -34,7 +34,7 @@ defmodule Scrapper.MixProject do
       {:sweet_xml, "~> 0.6"},
       {:broadway_rabbitmq, "~> 0.7"},
       {:amqp, "~> 3.3"},
-      {:sweet_xml, "~> 0.7.3"}
+      {:lol_analytics, in_umbrella: true}
       # {:dep_from_hexpm, "~> 0.3.0"},
       # {:dep_from_git, git: "https://github.com/elixir-lang/my_dep.git", tag: "0.1.0"},
       # {:sibling_app_in_umbrella, in_umbrella: true}

config/config.exs

@@ -14,6 +14,8 @@ config :lol_analytics,
   namespace: LoLAnalytics,
   ecto_repos: [LoLAnalytics.Repo]
 
+config :lol_analytics, LoLAnalytics.Repo, loggers: [], log: false
+
 config :lol_analytics_web,
   namespace: LoLAnalyticsWeb,
   ecto_repos: [LoLAnalytics.Repo],