Sideload lock state on installations (#1211)

apps/core/lib/core/schema/chart_installation.ex

@@ -4,6 +4,7 @@ defmodule Core.Schema.ChartInstallation do
 
   schema "chart_installations" do
     field :locked, :boolean, default: false
+    field :synced, :boolean, default: false
 
     belongs_to :installation, Installation
     belongs_to :chart,        Chart
@@ -80,7 +81,7 @@ defmodule Core.Schema.ChartInstallation do
   def preload(query \\ __MODULE__, preloads),
     do: from(cv in query, preload: ^preloads)
 
-  @valid ~w(installation_id chart_id version_id)a
+  @valid ~w(installation_id chart_id version_id synced)a
 
   def changeset(model, attrs \\ %{}) do
     model

apps/core/lib/core/schema/installation.ex

@@ -1,6 +1,6 @@
 defmodule Core.Schema.Installation do
   use Piazza.Ecto.Schema
-  alias Core.Schema.{Repository, User, Subscription, OIDCProvider}
+  alias Core.Schema.{Repository, User, Subscription, OIDCProvider, TerraformInstallation, ChartInstallation}
 
   defmodule Policy do
     use Piazza.Ecto.Schema
@@ -34,9 +34,30 @@ defmodule Core.Schema.Installation do
     has_one :subscription,  Subscription
     has_one :oidc_provider, OIDCProvider
 
+    has_many :terraform_installations, TerraformInstallation
+    has_many :chart_installations, ChartInstallation
+
     timestamps()
   end
 
+  def locks(query \\ __MODULE__) do
+    from(i in query,
+      left_join: ti in assoc(i, :terraform_installations),
+      left_join: ci in assoc(i, :chart_installations),
+      where: ti.locked or ci.locked,
+      select: %{id: i.id, locked: coalesce(ti.id, ci.id)}
+    )
+  end
+
+  def unsynced(query \\ __MODULE__) do
+    from(i in query,
+      left_join: ti in assoc(i, :terraform_installations),
+      left_join: ci in assoc(i, :chart_installations),
+      where: (not is_nil(ti.id) and not ti.synced) or (not is_nil(ci.id) and not ci.synced),
+      select: %{id: i.id, synced: coalesce(ti.id, ci.id)}
+    )
+  end
+
   def ordered(query \\ __MODULE__, order \\ [desc: :inserted_at]),
     do: from(i in query, order_by: ^order)
 

apps/core/lib/core/schema/terraform_installation.ex

@@ -4,6 +4,7 @@ defmodule Core.Schema.TerraformInstallation do
 
   schema "terraform_installations" do
     field :locked, :boolean, default: false
+    field :synced, :boolean, default: false
 
     belongs_to :installation, Installation
     belongs_to :terraform, Terraform
@@ -79,7 +80,7 @@ defmodule Core.Schema.TerraformInstallation do
     from(ti in query, preload: ^preloads)
   end
 
-  @valid ~w(installation_id terraform_id version_id)a
+  @valid ~w(installation_id terraform_id version_id synced)a
 
   def changeset(model, attrs \\ %{}) do
     model

apps/core/lib/core/services/repositories.ex

@@ -196,6 +196,31 @@ defmodule Core.Services.Repositories do
     |> Core.Repo.update()
   end
 
+  @doc """
+  Marks all current installations for a repository as having been synced, to be run during `plural deploy`
+  """
+  @spec synced(Repository.t, User.t) :: {:ok, map} | error
+  def synced(%Repository{id: id}, %User{id: user_id}) do
+    get_installation(user_id, id)
+    |> synced()
+  end
+
+  @spec synced(Installation.t) :: {:ok, map} | error
+  def synced(%Installation{id: id}) do
+    start_transaction()
+    |> add_operation(:tf, fn _ ->
+      TerraformInstallation.for_installation(id)
+      |> Core.Repo.update_all(set: [synced: true])
+      |> ok()
+    end)
+    |> add_operation(:helm, fn _ ->
+      ChartInstallation.for_installation(id)
+      |> Core.Repo.update_all(set: [synced: true])
+      |> ok()
+    end)
+    |> execute()
+  end
+
   @doc """
   Returns the list of docker accesses available for `user` against the
   given repository

apps/core/lib/core/services/upgrades.ex

@@ -274,7 +274,7 @@ defmodule Core.Services.Upgrades do
     |> add_operation(:lock, fn _ -> Rollouts.lock_installation(version, inst) end)
     |> add_operation(:inst, fn %{lock: inst} ->
       inst
-      |> Ecto.Changeset.change(%{version_id: version.id})
+      |> Ecto.Changeset.change(%{version_id: version.id, synced: false})
       |> Core.Repo.update()
       |> when_ok(&Core.Repo.preload(&1, [:installation]))
     end)

apps/core/priv/repo/migrations/20230823143301_add_installation_synced.exs

@@ -0,0 +1,13 @@
+defmodule Core.Repo.Migrations.AddInstallationSynced do
+  use Ecto.Migration
+
+  def change do
+    alter table(:chart_installations) do
+      add :synced, :boolean
+    end
+
+    alter table(:terraform_installations) do
+      add :synced, :boolean
+    end
+  end
+end

apps/core/test/services/repositories_test.exs

@@ -764,6 +764,17 @@ defmodule Core.Services.RepositoriesTest do
     end
   end
 
+  describe "#synced/1" do
+    test "it can mark module installations as synced" do
+      inst = insert(:installation)
+      ci = insert(:chart_installation, installation: inst)
+
+      {:ok, _} = Repositories.synced(inst)
+
+      assert refetch(ci).synced
+    end
+  end
+
   describe "#documentation/1" do
     test "it will find docs" do
       repo = insert(:repository, docs: %{file_name: "f", updated_at: nil})

apps/core/test/services/rollable/versions_test.exs

@@ -10,7 +10,8 @@ defmodule Core.Rollable.VersionsTest do
         insert(:chart_installation,
           installation: insert(:installation, auto_upgrade: true),
           chart: chart,
-          version: chart_version
+          version: chart_version,
+          synced: true
         )
       end
 
@@ -30,8 +31,11 @@ defmodule Core.Rollable.VersionsTest do
       assert rolled.status == :finished
       assert rolled.count == 3
 
-      for bumped <- auto_upgraded,
-        do: assert refetch(bumped).version_id == version.id
+      for bumped <- auto_upgraded do
+        bumped = refetch(bumped)
+        assert bumped.version_id == version.id
+        refute bumped.synced
+      end
 
       for ignore <- ignored,
         do: assert refetch(ignore).version_id == chart_version.id

apps/graphql/lib/graphql.ex

@@ -64,7 +64,9 @@ defmodule GraphQl do
     Cluster,
     Upgrade,
     GraphQl.InstallationLoader,
-    GraphQl.ShellLoader
+    GraphQl.ShellLoader,
+    GraphQl.LockLoader,
+    GraphQl.SyncLoader
   ]
 
   def context(ctx) do

apps/graphql/lib/graphql/resolvers/dataloaders.ex

@@ -18,6 +18,48 @@ defmodule GraphQl.InstallationLoader do
   end
 end
 
+defmodule GraphQl.LockLoader do
+  alias Core.Schema.Installation
+
+  def data(_) do
+    Dataloader.KV.new(&query/2, max_concurrency: 1)
+  end
+
+  def query(_, ids) do
+    locks = fetch_locks(ids)
+    Map.new(ids, & {&1, !!locks[&1]})
+  end
+
+  def fetch_locks(ids) do
+    MapSet.to_list(ids)
+    |> Installation.for_ids()
+    |> Installation.locks()
+    |> Core.Repo.all()
+    |> Map.new(& {&1.id, &1.locked})
+  end
+end
+
+defmodule GraphQl.SyncLoader do
+  alias Core.Schema.Installation
+
+  def data(_) do
+    Dataloader.KV.new(&query/2, max_concurrency: 1)
+  end
+
+  def query(_, ids) do
+    unsynced = fetch_unsynced(ids)
+    Map.new(ids, & {&1, !unsynced[&1]})
+  end
+
+  def fetch_unsynced(ids) do
+    MapSet.to_list(ids)
+    |> Installation.for_ids()
+    |> Installation.unsynced()
+    |> Core.Repo.all()
+    |> Map.new(& {&1.id, &1.synced})
+  end
+end
+
 defmodule GraphQl.ShellLoader do
   alias Core.Schema.CloudShell
 

apps/graphql/lib/graphql/resolvers/repository.ex

@@ -231,6 +231,15 @@ defmodule GraphQl.Resolvers.Repository do
     Repositories.release_apply_lock(attrs, repo.id, user)
   end
 
+  def synced(%{repository: name}, %{context: %{current_user: user}}) do
+    Repositories.get_repository_by_name!(name)
+    |> Repositories.synced(user)
+    |> case do
+      {:ok, _} -> {:ok, true}
+      error -> error
+    end
+  end
+
   def generate_scaffold(%{application: app} = ctx, _) do
     Core.Services.Scaffolds.generate(app, ctx)
     |> Enum.map(fn {file, content} ->

apps/graphql/lib/graphql/schema/repository.ex

@@ -8,6 +8,7 @@ defmodule GraphQl.Schema.Repository do
     Tag,
     Account
   }
+  alias GraphQl.{LockLoader, SyncLoader}
 
   ### INPUTS
 
@@ -147,6 +148,15 @@ defmodule GraphQl.Schema.Repository do
       _, _, _ -> {:ok, Core.conf(:acme_secret)}
     end
 
+    field :locked, :boolean, resolve: fn
+      %{id: id}, _, %{context: %{loader: loader}} ->
+        manual_dataloader(loader, LockLoader, :ids, id)
+    end
+
+    field :synced, :boolean, resolve: fn
+      %{id: id}, _, %{context: %{loader: loader}} ->
+        manual_dataloader(loader, SyncLoader, :ids, id)
+    end
 
     field :license, :string, resolve: fn
       installation, _, _ -> Core.Services.Repositories.generate_license(installation)
@@ -522,5 +532,12 @@ defmodule GraphQl.Schema.Repository do
 
       safe_resolve &Repository.release_apply_lock/2
     end
+
+    field :synced, :boolean do
+      middleware Authenticated
+      arg :repository, non_null(:string)
+
+      safe_resolve &Repository.synced/2
+    end
   end
 end

apps/graphql/test/queries/repository_queries_test.exs

@@ -105,20 +105,33 @@ defmodule GraphQl.RepositoryQueriesTest do
 
     test "It can list repositories installed by a user" do
       user = insert(:user)
-      installations = insert_list(3, :installation, user: user)
+      [inst | _ ] = installations = insert_list(3, :installation, user: user)
+      insert(:chart_installation, installation: inst, locked: true)
       insert(:repository)
 
       {:ok, %{data: %{"repositories" => repos}}} = run_query("""
         query {
           repositories(installed: true, first: 5) {
-            edges { node { id } }
+            edges {
+              node {
+                id
+                installation { id locked synced }
+              }
+            }
           }
         }
       """, %{}, %{current_user: user})
 
       found_repos = from_connection(repos)
       assert Enum.map(installations, & &1.repository)
              |> ids_equal(found_repos)
+
+      found = Enum.find(found_repos, & &1["installation"]["id"] == inst.id)
+      assert found["installation"]["locked"]
+      refute found["installation"]["synced"]
+
+      refute Enum.reject(found_repos, & &1["installation"]["id"] == inst.id)
+             |> Enum.any?(& &1["installation"]["locked"])
     end
 
     test "It can list repositories not installed by a user" do
@@ -571,6 +584,21 @@ defmodule GraphQl.RepositoryQueriesTest do
     end
   end
 
+  describe "synced" do
+    test "it can mark stuff as synced" do
+      inst = insert(:installation)
+      ci = insert(:chart_installation, installation: inst)
+
+      {:ok, %{data: %{"synced" => true}}} = run_query("""
+        mutation Synced($repo: String!) {
+          synced(repository: $repo)
+        }
+      """, %{"repo" => inst.repository.name}, %{current_user: inst.user})
+
+      assert refetch(ci).synced
+    end
+  end
+
   describe "scaffold" do
     test "it won't explode" do
       {:ok, %{data: %{"scaffold" => _}}} = run_query("""

schema/schema.graphql

@@ -1802,6 +1802,8 @@ type RootMutationType {
 
   releaseLock(repository: String!, attributes: LockAttributes!): ApplyLock
 
+  synced(repository: String!): Boolean
+
   createRecipe(repositoryName: String, repositoryId: String, attributes: RecipeAttributes!): Recipe
 
   deleteRecipe(id: ID!): Recipe
@@ -2049,6 +2051,10 @@ type Installation {
 
   acmeSecret: String
 
+  locked: Boolean
+
+  synced: Boolean
+
   license: String
 
   insertedAt: DateTime

www/src/generated/graphql.ts

@@ -1258,6 +1258,7 @@ export type Installation = {
   license?: Maybe<Scalars['String']['output']>;
   /** The license key for the application. */
   licenseKey?: Maybe<Scalars['String']['output']>;
+  locked?: Maybe<Scalars['Boolean']['output']>;
   /** The OIDC provider for the application. */
   oidcProvider?: Maybe<OidcProvider>;
   /** The last ping time of an installed application. */
@@ -1266,6 +1267,7 @@ export type Installation = {
   repository?: Maybe<Repository>;
   /** The subscription for the application. */
   subscription?: Maybe<RepositorySubscription>;
+  synced?: Maybe<Scalars['Boolean']['output']>;
   /** The tag to track for auto upgrades. */
   trackTag: Scalars['String']['output'];
   updatedAt?: Maybe<Scalars['DateTime']['output']>;
@@ -2775,6 +2777,7 @@ export type RootMutationType = {
   signup?: Maybe<User>;
   ssoCallback?: Maybe<User>;
   stopShell?: Maybe<Scalars['Boolean']['output']>;
+  synced?: Maybe<Scalars['Boolean']['output']>;
   transferDemoProject?: Maybe<DemoProject>;
   /** transfers ownership of a cluster to a service account */
   transferOwnership?: Maybe<Cluster>;
@@ -3376,6 +3379,11 @@ export type RootMutationTypeSsoCallbackArgs = {
 };
 
 
+export type RootMutationTypeSyncedArgs = {
+  repository: Scalars['String']['input'];
+};
+
+
 export type RootMutationTypeTransferDemoProjectArgs = {
   organizationId: Scalars['String']['input'];
 };