Support sandboxes in collections

CHANGELOG.md

@@ -17,6 +17,14 @@ and this project adheres to
 
 ### Added
 
+- Support collections in sandboxes. Collection names are now scoped per project,
+  empty collections are cloned into a sandbox on provision, and collection names
+  (not data) are synchronised when a sandbox is merged back into its parent.
+  Adds a v2 collections API at `/collections/:project_id/:name` selected via the
+  `x-api-version: 2` header. V1 continues to work and returns 409 when a name is
+  ambiguous across projects.
+  [#3548](https://github.com/OpenFn/lightning/issues/3548)
+
 ### Changed
 
 - Bump `@openfn/ws-worker` from

lib/lightning/collections.ex

@@ -53,9 +53,19 @@ defmodule Lightning.Collections do
   end
 
   @spec get_collection(String.t()) ::
-          {:ok, Collection.t()} | {:error, :not_found}
+          {:ok, Collection.t()} | {:error, :not_found} | {:error, :conflict}
   def get_collection(name) do
-    case Repo.get_by(Collection, name: name) do
+    case Repo.all(from c in Collection, where: c.name == ^name) do
+      [] -> {:error, :not_found}
+      [collection] -> {:ok, collection}
+      [_ | _] -> {:error, :conflict}
+    end
+  end
+
+  @spec get_collection(Ecto.UUID.t(), String.t()) ::
+          {:ok, Collection.t()} | {:error, :not_found}
+  def get_collection(project_id, name) do
+    case Repo.get_by(Collection, project_id: project_id, name: name) do
       nil -> {:error, :not_found}
       collection -> {:ok, collection}
     end

lib/lightning/collections/collection.ex

@@ -40,7 +40,8 @@ defmodule Lightning.Collections.Collection do
     |> validate_format(:name, ~r/^[a-z0-9]+([\-_.][a-z0-9]+)*$/,
       message: "Collection name must be URL safe"
     )
-    |> unique_constraint([:name],
+    |> unique_constraint(:name,
+      name: :collections_project_id_name_index,
       message: "A collection with this name already exists"
     )
   end
@@ -50,7 +51,8 @@ defmodule Lightning.Collections.Collection do
     |> validate_format(:name, ~r/^[a-z0-9]+([\-_.][a-z0-9]+)*$/,
       message: "Collection name must be URL safe"
     )
-    |> unique_constraint([:name],
+    |> unique_constraint(:name,
+      name: :collections_project_id_name_index,
       message: "A collection with this name already exists"
     )
   end

lib/lightning/projects/sandboxes.ex

@@ -18,6 +18,7 @@ defmodule Lightning.Projects.Sandboxes do
   ## Operations
 
   * `provision/3` - Create a new sandbox from a parent project
+  * `merge/4` - Merge a sandbox into its target (workflows + collections)
   * `update_sandbox/3` - Update sandbox name, color, or environment
   * `delete_sandbox/2` - Delete a sandbox and all its descendants
 
@@ -36,10 +37,14 @@ defmodule Lightning.Projects.Sandboxes do
   import Ecto.Query
 
   alias Lightning.Accounts.User
+  alias Lightning.Collections
+  alias Lightning.Collections.Collection
   alias Lightning.Credentials.KeychainCredential
   alias Lightning.Policies.Permissions
+  alias Lightning.Projects.MergeProjects
   alias Lightning.Projects.Project
   alias Lightning.Projects.ProjectCredential
+  alias Lightning.Projects.Provisioner
   alias Lightning.Projects.SandboxPromExPlugin
   alias Lightning.Repo
   alias Lightning.Workflows
@@ -121,6 +126,42 @@ defmodule Lightning.Projects.Sandboxes do
     end
   end
 
+  @doc """
+  Merges a sandbox into its target project.
+
+  Applies the sandbox's workflow configuration to the target via the
+  provisioner, then synchronises collection names. Collection data is
+  never copied.
+
+  ## Parameters
+  * `source` - The sandbox project being merged
+  * `target` - The project receiving the merge
+  * `actor` - The user performing the merge
+  * `opts` - Merge options (`:selected_workflow_ids`, `:deleted_target_workflow_ids`)
+
+  ## Returns
+  * `{:ok, updated_target}` - Merge and collection sync succeeded
+  * `{:error, reason}` - Workflow merge or collection sync failed
+  """
+  @spec merge(Project.t(), Project.t(), User.t(), map()) ::
+          {:ok, Project.t()} | {:error, term()}
+  def merge(
+        %Project{} = source,
+        %Project{} = target,
+        %User{} = actor,
+        opts \\ %{}
+      ) do
+    merge_doc = MergeProjects.merge_project(source, target, opts)
+
+    with {:ok, updated_target} <-
+           Provisioner.import_document(target, actor, merge_doc,
+             allow_stale: true
+           ),
+         {:ok, _} <- sync_collections(source, target) do
+      {:ok, updated_target}
+    end
+  end
+
   @doc """
   Updates a sandbox project's basic attributes.
 
@@ -566,6 +607,87 @@ defmodule Lightning.Projects.Sandboxes do
     |> copy_workflow_version_history(sandbox.workflow_id_mapping)
     |> create_initial_workflow_snapshots()
     |> copy_selected_dataclips(parent.id, Map.get(original_attrs, :dataclip_ids))
+    |> clone_collections_from_parent(parent)
+  end
+
+  defp clone_collections_from_parent(sandbox, parent) do
+    parent_names = parent |> Collections.list_project_collections() |> names()
+    insert_empty_collections(sandbox.id, parent_names)
+    sandbox
+  end
+
+  @doc """
+  Synchronises collection names from a sandbox to its merge target.
+
+  After a successful merge, this brings the target's set of collections in
+  line with the sandbox's:
+
+    * Collections present in the sandbox but missing from the target are
+      created (empty) in the target.
+    * Collections present in the target but missing from the sandbox are
+      deleted from the target, along with all their items.
+
+  **Collection data is never copied or merged.** Only the set of collection
+  names is synchronised, mirroring the sandbox-is-for-configuration model.
+
+  The create and delete operations run in a single transaction; a failure
+  leaves the target's collections unchanged.
+  """
+  @spec sync_collections(Project.t(), Project.t()) ::
+          {:ok, %{created: non_neg_integer(), deleted: non_neg_integer()}}
+          | {:error, term()}
+  def sync_collections(%Project{} = source, %Project{} = target) do
+    source_names = source |> Collections.list_project_collections() |> names()
+
+    target_collections = Collections.list_project_collections(target)
+    target_names = names(target_collections)
+
+    to_create = MapSet.difference(source_names, target_names)
+
+    names_to_delete = MapSet.difference(target_names, source_names)
+
+    to_delete_ids =
+      for c <- target_collections,
+          c.name in names_to_delete,
+          do: c.id
+
+    Repo.transaction(fn ->
+      {created, _} = insert_empty_collections(target.id, to_create)
+      {deleted, _} = delete_collections(to_delete_ids)
+      %{created: created, deleted: deleted}
+    end)
+  end
+
+  defp names(collections), do: MapSet.new(collections, & &1.name)
+
+  defp insert_empty_collections(project_id, names) do
+    if Enum.empty?(names) do
+      {0, nil}
+    else
+      now = DateTime.utc_now() |> DateTime.truncate(:second)
+
+      rows =
+        Enum.map(names, fn name ->
+          %{
+            id: Ecto.UUID.generate(),
+            name: name,
+            project_id: project_id,
+            byte_size_sum: 0,
+            inserted_at: now,
+            updated_at: now
+          }
+        end)
+
+      # on_conflict: :nothing handles the rare case where two concurrent
+      # merges into the same target both try to create the same collection.
+      Repo.insert_all(Collection, rows, on_conflict: :nothing)
+    end
+  end
+
+  defp delete_collections([]), do: {0, nil}
+
+  defp delete_collections(ids) do
+    Repo.delete_all(from c in Collection, where: c.id in ^ids)
   end
 
   defp copy_workflow_version_history(sandbox, workflow_id_mapping) do