Add subscriptions projection

lib/tronto/monitoring/domain/hosts/commands/update_sles_subscriptions.ex

@@ -0,0 +1,21 @@
+defmodule Tronto.Monitoring.Domain.Commands.UpdateSlesSubscriptions do
+  @moduledoc """
+  Update data relative to subscriptions.
+  """
+
+  use TypedStruct
+  use Domo
+
+  alias Tronto.Monitoring.Domain.SlesSubscription
+
+  typedstruct do
+    @typedoc "UpdateSubscriptions command"
+
+    field :subscriptions, [SlesSubscription.t()], enforce: true
+    field :host_id, String.t(), enforce: true
+  end
+
+  use Vex.Struct
+
+  validates :host_id, uuid: true
+end

lib/tronto/monitoring/domain/hosts/events/sles_subscriptions_updated.ex

@@ -0,0 +1,17 @@
+defmodule Tronto.Monitoring.Domain.Events.SlesSubscriptionsUpdated do
+  @moduledoc """
+  Subscriptions updated event
+  """
+
+  alias Tronto.Monitoring.Domain.SlesSubscription
+
+  use TypedStruct
+
+  @derive Jason.Encoder
+  typedstruct do
+    @typedoc "SubscriptionsUpdated event"
+
+    field :host_id, String.t(), enforce: true
+    field :subscriptions, [SlesSubscription.t()], enforce: true
+  end
+end

lib/tronto/monitoring/domain/hosts/host.ex

@@ -3,18 +3,22 @@ defmodule Tronto.Monitoring.Domain.Host do
 
   alias Tronto.Monitoring.Domain.Host
 
+  alias Tronto.Monitoring.Domain.SlesSubscription
+
   alias Tronto.Monitoring.Domain.Commands.{
     RegisterHost,
     UpdateHeartbeat,
-    UpdateProvider
+    UpdateProvider,
+    UpdateSlesSubscriptions
   }
 
   alias Tronto.Monitoring.Domain.Events.{
     HeartbeatFailed,
     HeartbeatSucceded,
     HostDetailsUpdated,
     HostRegistered,
-    ProviderUpdated
+    ProviderUpdated,
+    SlesSubscriptionsUpdated
   }
 
   defstruct [
@@ -23,7 +27,8 @@ defmodule Tronto.Monitoring.Domain.Host do
     :ip_addresses,
     :agent_version,
     :provider,
-    :heartbeat
+    :heartbeat,
+    :subscriptions
   ]
 
   @type t :: %__MODULE__{
@@ -32,6 +37,7 @@ defmodule Tronto.Monitoring.Domain.Host do
           ip_addresses: [String.t()],
           agent_version: String.t(),
           provider: String.t(),
+          subscriptions: [SlesSubscription.t()],
           heartbeat: :passing | :critical | :unknown
         }
 
@@ -133,6 +139,13 @@ defmodule Tronto.Monitoring.Domain.Host do
     []
   end
 
+  def execute(
+        %Host{host_id: nil},
+        %UpdateSlesSubscriptions{}
+      ) do
+    {:error, :host_not_registered}
+  end
+
   def execute(
         %Host{},
         %UpdateProvider{host_id: host_id, provider: provider}
@@ -143,6 +156,22 @@ defmodule Tronto.Monitoring.Domain.Host do
     }
   end
 
+  def execute(%Host{subscriptions: subscriptions}, %UpdateSlesSubscriptions{
+        subscriptions: subscriptions
+      }) do
+    []
+  end
+
+  def execute(%Host{}, %UpdateSlesSubscriptions{
+        host_id: host_id,
+        subscriptions: subscriptions
+      }) do
+    %SlesSubscriptionsUpdated{
+      host_id: host_id,
+      subscriptions: subscriptions
+    }
+  end
+
   def apply(
         %Host{} = host,
         %HostRegistered{
@@ -210,4 +239,10 @@ defmodule Tronto.Monitoring.Domain.Host do
       | provider: provider
     }
   end
+
+  def apply(%Host{} = host, %SlesSubscriptionsUpdated{
+        subscriptions: subscriptions
+      }) do
+    %Host{host | subscriptions: subscriptions}
+  end
 end

lib/tronto/monitoring/domain/value_objects/sles_subscription.ex

@@ -0,0 +1,23 @@
+defmodule Tronto.Monitoring.Domain.SlesSubscription do
+  @moduledoc """
+  SLES subscriptions value object
+  """
+
+  use TypedStruct
+  use Domo
+
+  @derive Jason.Encoder
+  typedstruct do
+    @typedoc "SlesSubscription value object"
+
+    field :host_id, String.t(), enforce: true
+    field :identifier, String.t(), enforce: true
+    field :version, String.t(), enforce: true
+    field :arch, String.t(), enforce: true
+    field :status, String.t(), enforce: true
+    field :subscription_status, String.t()
+    field :type, String.t()
+    field :starts_at, String.t()
+    field :expires_at, String.t()
+  end
+end

lib/tronto/monitoring/integration/discovery.ex

@@ -9,9 +9,12 @@ defmodule Tronto.Monitoring.Integration.Discovery do
   alias Tronto.Monitoring.Domain.Commands.{
     RegisterCluster,
     RegisterHost,
-    UpdateProvider
+    UpdateProvider,
+    UpdateSlesSubscriptions
   }
 
+  alias Tronto.Monitoring.Domain.SlesSubscription
+
   @spec handle_discovery_event(map) :: {:error, any} | {:ok, command}
   def handle_discovery_event(%{
         "discovery_type" => "host_discovery",
@@ -67,6 +70,17 @@ defmodule Tronto.Monitoring.Integration.Discovery do
     )
   end
 
+  def handle_discovery_event(%{
+        "discovery_type" => "subscription_discovery",
+        "agent_id" => agent_id,
+        "payload" => payload
+      }) do
+    subscriptions =
+      Enum.map(payload, fn subscription -> parse_subscription_data(agent_id, subscription) end)
+
+    UpdateSlesSubscriptions.new(host_id: agent_id, subscriptions: subscriptions)
+  end
+
   def handle_discovery_event(_) do
     {:error, :invalid_payload}
   end
@@ -123,4 +137,42 @@ defmodule Tronto.Monitoring.Integration.Discovery do
         nil
     end)
   end
+
+  defp parse_subscription_data(host_id, %{
+         "arch" => arch,
+         "expires_at" => expires_at,
+         "identifier" => identifier,
+         "starts_at" => starts_at,
+         "status" => status,
+         "subscription_status" => subscription_status,
+         "type" => type,
+         "version" => version
+       }) do
+    SlesSubscription.new!(
+      host_id: host_id,
+      arch: arch,
+      expires_at: expires_at,
+      identifier: identifier,
+      starts_at: starts_at,
+      status: status,
+      subscription_status: subscription_status,
+      type: type,
+      version: version
+    )
+  end
+
+  defp parse_subscription_data(host_id, %{
+         "arch" => arch,
+         "identifier" => identifier,
+         "status" => status,
+         "version" => version
+       }) do
+    SlesSubscription.new!(
+      host_id: host_id,
+      arch: arch,
+      identifier: identifier,
+      status: status,
+      version: version
+    )
+  end
 end

lib/tronto/monitoring/projectors/projectors_supervisor.ex

@@ -12,7 +12,8 @@ defmodule Tronto.Monitoring.ProjectorsSupervisor do
     children = [
       Tronto.Monitoring.ClusterProjector,
       Tronto.Monitoring.HostProjector,
-      Tronto.Monitoring.CheckResultProjector
+      Tronto.Monitoring.CheckResultProjector,
+      Tronto.Monitoring.SlesSubscriptionsProjector
     ]
 
     Supervisor.init(children, strategy: :one_for_one)

lib/tronto/monitoring/projectors/sles_subscriptions_projector.ex

@@ -0,0 +1,40 @@
+defmodule Tronto.Monitoring.SlesSubscriptionsProjector do
+  @moduledoc """
+  Check result projector
+  """
+
+  use Commanded.Projections.Ecto,
+    application: Tronto.Commanded,
+    repo: Tronto.Repo,
+    name: "sles_subscription_projector"
+
+  import Ecto.Query
+
+  alias Tronto.Monitoring.Domain.Events.SlesSubscriptionsUpdated
+
+  alias Tronto.Monitoring.SlesSubscriptionReadModel
+
+  project(
+    %SlesSubscriptionsUpdated{host_id: host_id, subscriptions: subscriptions},
+    fn multi ->
+      multi =
+        Ecto.Multi.delete_all(
+          multi,
+          :delete_old_sles_subscriptions,
+          from(s in SlesSubscriptionReadModel, where: s.host_id == ^host_id)
+        )
+
+      subscriptions
+      |> Enum.map(fn subscription ->
+        SlesSubscriptionReadModel.changeset(
+          %SlesSubscriptionReadModel{},
+          subscription
+        )
+      end)
+      |> Enum.reduce(multi, fn %{changes: %{host_id: host_id, identifier: identifier}} = changeset,
+                               acc ->
+        Ecto.Multi.insert(acc, "#{host_id}_#{identifier}", changeset)
+      end)
+    end
+  )
+end

lib/tronto/monitoring/read_models/sles_subscription_read_model.ex

@@ -0,0 +1,35 @@
+defmodule Tronto.Monitoring.SlesSubscriptionReadModel do
+  @moduledoc """
+  SLES subscriptions read model
+  """
+
+  use Ecto.Schema
+
+  import Ecto.Changeset
+
+  alias Tronto.Monitoring.HostReadModel
+
+  @type t :: %__MODULE__{}
+
+  @derive {Jason.Encoder, except: [:__meta__, :__struct__]}
+  @primary_key false
+  schema "sles_subscriptions" do
+    field :host_id, Ecto.UUID, primary_key: true
+    field :identifier, :string, primary_key: true
+    field :version, :string
+    field :arch, :string
+    field :status, :string
+    field :subscription_status, :string
+    field :type, :string
+    field :starts_at, :string
+    field :expires_at, :string
+
+    has_one :host, HostReadModel, references: :host_id, foreign_key: :id
+    timestamps()
+  end
+
+  @spec changeset(t() | Ecto.Changeset.t(), map) :: Ecto.Changeset.t()
+  def changeset(sles_subscription, attrs) do
+    cast(sles_subscription, attrs, __MODULE__.__schema__(:fields))
+  end
+end

lib/tronto/router.ex

@@ -15,13 +15,14 @@ defmodule Tronto.Router do
     SelectChecks,
     StoreChecksResults,
     UpdateHeartbeat,
-    UpdateProvider
+    UpdateProvider,
+    UpdateSlesSubscriptions
   }
 
   middleware Validate
 
   identify Host, by: :host_id
-  dispatch [RegisterHost, UpdateHeartbeat, UpdateProvider], to: Host
+  dispatch [RegisterHost, UpdateHeartbeat, UpdateProvider, UpdateSlesSubscriptions], to: Host
 
   identify Cluster, by: :cluster_id
 

priv/repo/migrations/20220302154246_create_sles_subscription_read_model.exs

@@ -0,0 +1,21 @@
+defmodule Tronto.Repo.Migrations.CreateSlesSubscriptionReadModel do
+  use Ecto.Migration
+
+  def change do
+    create table(:sles_subscriptions, primary_key: false) do
+      add :host_id, :uuid, primary_key: true
+      add :identifier, :string, primary_key: true
+      add :version, :string
+      add :arch, :string
+      add :status, :string
+      add :subscription_status, :string
+      add :type, :string
+      add :starts_at, :string
+      add :expires_at, :string
+
+      timestamps
+    end
+
+    create index(:sles_subscriptions, [:host_id])
+  end
+end