Sku Demand (#11)

* work

* work drop

* update genserver and kit demand calculations

* create component genserver to track demand

* update genservers to handle demands and availability

* add more demand warmup logic

* fix spell check

* remove need for EctoEnum

* clean up grpc testing for higher test coverage

* fix formatting

* update configuration values

* update warmup process

* update logging information and take down more debounce

* add timeout for sku start process

* more updates to server and grpc

.formatter.exs

@@ -1,5 +1,5 @@
 [
-  import_deps: [:ecto, :ecto_enum, :ecto_sql, :grpc],
+  import_deps: [:ecto, :ecto_sql, :grpc],
   inputs: [
     "*.{ex,exs}",
     "{config,lib,test}/**/*.{ex,exs}"

README.md

@@ -1,3 +1,106 @@
-# Warehouse
+<div align="center">
+  <h1>Warehouse</h1>
+  <h3>An inventory tracking microservice</h3>
+  <br>
+  <br>
+</div>
 
-An inventory microservice
+---
+
+> **NOTE**: This micro service is not fully written yet, and includes references
+> to database records, processes, and workflows that are not yet implemented in
+> here.
+
+This repository contains the code that System76 uses to manage it's warehouse of
+computer parts. It is responsible for:
+
+- Creating POs with vendors to receive new parts
+- Manage receiving new parts from vendors
+- Manage kitting and the relationship between what we have in inventory and what
+  is requested from the e-commerce orders.
+- Calculating and tracking the demand for different SKUs in our system (available, back ordered, etc)
+
+## Communication
+
+This micro service works very closely with (and is dependent on)
+[the Assembly service](https://github.com/system76/assembly). The assembly
+service is responsible for tracking build details. They have a relationship like
+so:
+
+```
+Assembly ------------------------------------------------------------> Warehouse
+
+This is a gRPC request from Assembly to Warehouse to determine the
+`Warehouse.Schema.Component` quantity available. This is used to determine if a
+`Assembly.Schemas.Build` has all of the needed parts in stock to build. A
+similar RabbitMQ message is broadcasted when that quantity changes.
+
+Assembly <------------------------------------------------------------ Warehouse
+
+This is a gRPC request from Warehouse to Assembly to determine the demand of
+`Warehouse.Schema.Component`. This allows Warehouse to determine the back order
+status of a `Warehouse.Schema.Sku` and the quantity we need to order. A similar
+RabbitMQ message is broadcasted when this quantity changes.
+```
+
+## Schemas
+
+This micro service as a couple of schemas it uses, but two stand out as the
+cornerstones.
+
+```
+        component_1             component_2              <- `Warehouse.Schemas.Component`
+       /           \           /           \
+      /             \         /             \            <- `Warehouse.Schemas.Kit`
+     /               \       /               \
+sku_1                 sku_two                 sku_two    <- `Warehouse.Schemas.Sku`
+```
+
+The `Warehouse.Schemas.Component` schema connects our e-ecommerce platform and
+assembly system to inventory. Anything that gets sold is a component. When you
+purchase a computer, every component selected (and some hidden components), are
+added to a build. These components are represented as a very basic, customer
+facing names like `NVIDIA RTX 3080`.
+
+The `Warehouse.Schemas.Sku` schema is our lower level inventory system. This
+is a much more specific product that we buy from vendors, like
+`MSI GeForce RTX 3080 GAMING X TRIO 10GB`, or `G3080GXT10`.
+
+The `Warehouse.Schemas.Kit` schema is how we combine the other two schemas.
+Every `Warehouse.Schemas.Component` can be fulfilled by any selected
+`Warehouse.Schemas.Sku`, and most `Warehouse.Schemas.Sku`s can be used by
+multiple `Warehouse.Schemas.Component`s. This comes into play with more complex
+configurations like memory. A pseudo example of this:
+
+```
+%Component{id: 1, name: "32 GB DDR4 @ 3200 MHz Desktop Memory"}
+
+%Kit{component_id: 1, sku_id: 1, quantity: 4}
+%Sku{sku_id: 1, name: "Kingston 8 GB DDR4 at 3200 MHz"}
+
+%Kit{component_id: 1, sku_id: 2, quantity: 4}
+%Sku{sku_id: 2, name: "Crucial 16 GB DDR4 at 3200 MHz"}
+
+%Kit{component_id: 1, sku_id: 3, quantity: 2}
+%Sku{sku_id: 3, name: "Kingston 16 GB DDR4 at 3200 MHz"}
+```
+
+## Setup
+
+First, make sure you are running the dependency services with `docker-compose`:
+
+```shell
+docker-compose up
+```
+
+Dependencies are managed via `mix`. In the repo, run:
+
+```shell
+mix deps.get
+```
+
+Then run this to test the project:
+
+```shell
+mix test
+```

config/config.exs

@@ -2,10 +2,8 @@ import Config
 
 config :warehouse,
   env: Mix.env(),
-  ecto_repos: [Warehouse.Repo]
-
-config :warehouse,
-  producer: {BroadwayRabbitMQ.Producer, queue: "", connection: []},
+  ecto_repos: [Warehouse.Repo],
+  events: Warehouse.Events,
   exluded_picking_locations: [
     # shipping
     208,
@@ -23,11 +21,13 @@ config :warehouse,
     400,
     # sarah's desk
     401
-  ]
+  ],
+  producer: {BroadwayRabbitMQ.Producer, queue: "", connection: []},
+  warmup: &Warehouse.warmup/0
 
 config :logger, :console,
   format: "$time $metadata[$level] $message\n",
-  metadata: [:request_id, :part_id, :build_id, :trace_id, :span_id, :resource],
+  metadata: [:request_id, :part_id, :build_id, :component_id, :sku_id, :trace_id, :span_id, :resource],
   level: :info
 
 config :logger_json, :backend,
@@ -41,6 +41,11 @@ config :ex_aws,
   secret_access_key: nil,
   region: nil
 
+config :warehouse, Warehouse.AssemblyServiceClient,
+  enabled?: false,
+  url: "",
+  ssl: false
+
 config :warehouse, Warehouse.Tracer,
   service: :warehouse,
   adapter: SpandexDatadog.Adapter,

config/releases.exs

@@ -40,4 +40,9 @@ config :amqp,
     events: [connection: :rabbitmq_conn]
   ]
 
+config :warehouse, Warehouse.AssemblyServiceClient,
+  enabled?: true,
+  url: warehouse_config["ASSEMBLY_SERVICE_URL"],
+  ssl: true
+
 config :warehouse, Warehouse.Tracer, env: warehouse_config["ENVIRONMENT"]

config/test.exs

@@ -1,7 +1,9 @@
 import Config
 
 config :warehouse,
-  producer: {Broadway.DummyProducer, []}
+  events: Warehouse.MockEvents,
+  producer: {Broadway.DummyProducer, []},
+  warmup: fn -> :ok end
 
 config :warehouse, Warehouse.Repo,
   username: "root",

lib/warehouse.ex

@@ -1,5 +1,35 @@
 defmodule Warehouse do
   @moduledoc """
-  Review all incoming orders for signs of fraud
+  The Warehouse microservice is responsible for handling different parts of
+  our inventory system, including:
+
+  - Creating POs with vendors to receive new parts
+  - Manage receiving new parts from vendors
+  - Manage kitting and the relationship between what we have in inventory and
+    what is requested from the e-commerce orders.
+  - Calculating and tracking the demand for different SKUs in our system
+    (available, back ordered, etc)
+
   """
+
+  @doc """
+  This starts all GenServers required for Warehouse to run.
+
+  ## Examples
+
+      iex> warmup()
+      :ok
+
+  """
+  def warmup() do
+    with :ok <- Warehouse.Component.warmup_components(),
+         :ok <- Warehouse.Sku.warmup_skus() do
+      Warehouse.AssemblyService.request_component_demands()
+      |> Stream.map(fn %{component_id: id, demand_quantity: demand} -> [id, demand] end)
+      |> Stream.each(&apply(Warehouse.Component, :update_component_demand, &1))
+      |> Stream.run()
+
+      :ok
+    end
+  end
 end

lib/warehouse/additive_map.ex

@@ -0,0 +1,84 @@
+defmodule Warehouse.AdditiveMap do
+  @moduledoc """
+  The `Warehouse.AdditiveMap` module represents a key value storage structure
+  where every value is a non_neg_integer. It is used to represent demand and
+  availability in the system due to it's helpful utility functions like `add/3`
+  and `merge/2`.
+  """
+
+  @type key :: String.t()
+  @type value :: non_neg_integer()
+
+  @type t :: %{required(key) => value}
+
+  @doc """
+  Adds demand to a demand map.
+
+  ## Examples
+
+      iex> add(%{"A" => 1}, "A", 2)
+      %{"A" => 3}
+
+      iex> add(%{}, "A", 1)
+      %{"A" => 1}
+
+  """
+  @spec add(t(), any(), value()) :: t()
+  def add(map, key, value) do
+    set(map, key, get(map, key) + value)
+  end
+
+  @doc """
+  Grabs the demand quantity for a given key.
+
+  ## Examples
+
+      iex> get(%{"A" => 1}, "A")
+      1
+
+      iex> get(%{}, "A")
+      0
+
+  """
+  @spec get(t(), any()) :: value()
+  def get(map, key) do
+    Map.get(map, to_string(key), 0)
+  end
+
+  @doc """
+  Merges two demand maps. This is very similar to `Map.merge/2` except it sums
+  up the values instead of overwrites.
+
+  ## Examples
+
+      iex> merge(%{"A" => 1, "B" => 2}, %{"A" => 4})
+      %{"A" => 5, "B" => 2}
+
+      iex> merge(%{"A" => 1, "B" => 2}, %{"C" => 3})
+      %{"A" => 1, "B" => 2, "C" => 3}
+
+  """
+  @spec merge(t(), t()) :: t()
+  def merge(one, two) do
+    Enum.reduce(two, one, fn {key, value}, map ->
+      add(map, key, value)
+    end)
+  end
+
+  @doc """
+  Sets the exact amount of demand in a demand map.
+
+  ## Examples
+
+      iex> set(%{"A" => 2}, "A", 4)
+      %{"A" => 4}
+
+      iex> set(%{}, "A", 2)
+      %{"A" => 2}
+
+  """
+  @spec set(t(), any(), value()) :: t()
+  def set(map, key, value) do
+    Map.put(map, to_string(key), value)
+  end
+end

lib/warehouse/application.ex

@@ -10,14 +10,33 @@ defmodule Warehouse.Application do
   def start(_type, _args) do
     children = [
       {SpandexDatadog.ApiServer, [http: HTTPoison, host: "127.0.0.1", batch_size: 20]},
+      {Task.Supervisor, name: Warehouse.TaskSupervisor},
+      {Registry, keys: :unique, name: Warehouse.ComponentRegistry},
+      {DynamicSupervisor, name: Warehouse.ComponentSupervisor, strategy: :one_for_one},
+      {Registry, keys: :unique, name: Warehouse.SkuRegistry},
+      {DynamicSupervisor, name: Warehouse.SkuSupervisor, strategy: :one_for_one},
       Warehouse.Repo,
       {GRPC.Server.Supervisor, {Warehouse.Endpoint, 50_051}},
       {Warehouse.Broadway, []}
     ]
 
+    children =
+      if Application.get_env(:warehouse, Warehouse.AssemblyServiceClient)[:enabled?],
+        do: children ++ [Warehouse.AssemblyServiceClient],
+        else: children
+
     Logger.info("Starting Warehouse")
 
     opts = [strategy: :one_for_one, name: Warehouse.Supervisor]
-    Supervisor.start_link(children, opts)
+
+    with {:ok, pid} <- Supervisor.start_link(children, opts) do
+      Task.Supervisor.async_nolink(Warehouse.TaskSupervisor, fn ->
+        :warehouse
+        |> Application.get_env(:warmup)
+        |> apply([])
+      end)
+
+      {:ok, pid}
+    end
   end
 end

lib/warehouse/assembly_service.ex

@@ -0,0 +1,27 @@
+defmodule Warehouse.AssemblyService do
+  @moduledoc """
+  Handles forming the request and parsing the response from the assembly
+  microservice gRPC server.
+  """
+
+  require Logger
+
+  alias Bottle.Assembly.V1.{ListComponentDemandsRequest, Stub}
+  alias Warehouse.AssemblyServiceClient
+
+  @spec request_component_demands() :: Enumerable.t()
+  def request_component_demands() do
+    request = ListComponentDemandsRequest.new(request_id: Bottle.RequestId.write(:queue))
+
+    with {:ok, channel} <- AssemblyServiceClient.channel(),
+         {:ok, stream} <- Stub.list_component_demands(channel, request) do
+      Stream.map(stream, &cast/1)
+    else
+      {:error, reason} ->
+        Logger.error("Unable to get component demand from assembly service", resource: inspect(reason))
+        Stream.cycle([])
+    end
+  end
+
+  defp cast({:ok, res}), do: res
+end