Merge branch 'main' into feature/belongs-to-resource-attribute-type

Author: zachdaniel

lib/ash_storage/changes/attach.ex

@@ -13,15 +13,16 @@ defmodule AshStorage.Changes.Attach do
   end
 
   @impl true
-  def change(changeset, opts, _context) do
+  def change(changeset, opts, context) do
     attachment_name = opts[:attachment_name]
+    context_opts = Ash.Context.to_opts(context)
 
     changeset
     |> Ash.Changeset.before_action(fn changeset ->
       record = changeset.data
       resource = record.__struct__
 
-      case do_attach(record, resource, attachment_name, changeset) do
+      case do_attach(record, resource, attachment_name, changeset, context_opts) do
         {:ok, attrs_to_write, attach_context} ->
           changeset
           |> Ash.Changeset.force_change_attributes(attrs_to_write)
@@ -33,15 +34,22 @@ defmodule AshStorage.Changes.Attach do
     end)
     |> Ash.Changeset.after_action(fn changeset, record ->
       case changeset.context[:__ash_storage_attach__] do
-        %{blob: blob, attachment_def: attachment_def, service_mod: service_mod, ctx: ctx} =
-            context ->
+        %{
+          blob: blob,
+          attachment_def: attachment_def,
+          service_mod: service_mod,
+          ctx: ctx,
+          context_opts: context_opts
+        } = attach_context ->
           resource = record.__struct__
 
-          with {:ok, _} <- maybe_replace_existing(record, attachment_def, service_mod, ctx),
-               {:ok, attachment} <- create_attachment(record, attachment_def, blob),
+          with {:ok, _} <-
+                 maybe_replace_existing(record, attachment_def, service_mod, ctx, context_opts),
+               {:ok, attachment} <-
+                 create_attachment(record, attachment_def, blob, context_opts),
                :ok <- run_eager_variants(blob, attachment_def, resource),
                {:ok, blob} <- store_oban_variants(blob, attachment_def, resource) do
-            if context[:has_oban_analyzers?] do
+            if attach_context[:has_oban_analyzers?] do
               AshOban.run_trigger(blob, :run_pending_analyzers)
             end
 
@@ -63,7 +71,7 @@ defmodule AshStorage.Changes.Attach do
     end)
   end
 
-  defp do_attach(record, resource, attachment_name, changeset) do
+  defp do_attach(record, resource, attachment_name, changeset, context_opts) do
     io = Ash.Changeset.get_argument(changeset, :io)
     filename = Ash.Changeset.get_argument(changeset, :filename)
 
@@ -77,7 +85,7 @@ defmodule AshStorage.Changes.Attach do
       ctx = build_context(service_opts, resource, attachment_def, changeset)
 
       with {:ok, blob} <-
-             upload_and_create_blob(resource, service_mod, ctx, io,
+             upload_and_create_blob(resource, service_mod, ctx, io, context_opts,
                filename: filename,
                content_type: content_type,
                metadata: metadata
@@ -89,6 +97,7 @@ defmodule AshStorage.Changes.Attach do
            attachment_def: attachment_def,
            service_mod: service_mod,
            ctx: ctx,
+           context_opts: context_opts,
            has_oban_analyzers?: has_oban_analyzers?(attachment_def)
          }}
       end
@@ -234,7 +243,7 @@ defmodule AshStorage.Changes.Attach do
     end
   end
 
-  defp upload_and_create_blob(resource, service_mod, ctx, io, opts) do
+  defp upload_and_create_blob(resource, service_mod, ctx, io, context_opts, opts) do
     filename = Keyword.fetch!(opts, :filename)
     content_type = Keyword.get(opts, :content_type, "application/octet-stream")
     metadata = Keyword.get(opts, :metadata, %{})
@@ -244,11 +253,10 @@ defmodule AshStorage.Changes.Attach do
     checksum = :crypto.hash(:md5, data) |> Base.encode64()
     byte_size = byte_size(data)
 
-    with :ok <- service_mod.upload(key, data, ctx) do
+    with {:ok, extra_blob_attrs} <- normalize_upload(service_mod.upload(key, data, ctx)) do
       blob_resource = Info.storage_blob_resource!(resource)
 
-      Ash.create(
-        blob_resource,
+      blob_attrs =
         %{
           key: key,
           filename: filename,
@@ -258,12 +266,17 @@ defmodule AshStorage.Changes.Attach do
           service_name: service_mod,
           service_opts: persistable_service_opts(service_mod, ctx.service_opts),
           metadata: metadata
-        },
-        action: :create
-      )
+        }
+        |> Map.merge(extra_blob_attrs)
+
+      Ash.create(blob_resource, blob_attrs, Keyword.merge(context_opts, action: :create))
     end
   end
 
+  defp normalize_upload(:ok), do: {:ok, %{}}
+  defp normalize_upload({:ok, attrs}) when is_map(attrs), do: {:ok, attrs}
+  defp normalize_upload({:error, _} = error), do: error
+
   # -- IO helpers --
 
   defp read_io(%Ash.Type.File{} = file) do
@@ -325,17 +338,24 @@ defmodule AshStorage.Changes.Attach do
   # -- Attachment helpers --
 
   # sobelow_skip ["DOS.BinToAtom"]
-  defp maybe_replace_existing(record, %{type: :one} = attachment_def, service_mod, ctx) do
-    case find_attachments(record, attachment_def) do
+  defp maybe_replace_existing(
+         record,
+         %{type: :one} = attachment_def,
+         service_mod,
+         ctx,
+         context_opts
+       ) do
+    case find_attachments(record, attachment_def, context_opts) do
       {:ok, []} -> {:ok, :noop}
-      {:ok, existing} -> purge_attachments(existing, service_mod, ctx)
+      {:ok, existing} -> purge_attachments(existing, service_mod, ctx, context_opts)
     end
   end
 
-  defp maybe_replace_existing(_record, %{type: :many}, _service_mod, _ctx), do: {:ok, :noop}
+  defp maybe_replace_existing(_record, %{type: :many}, _service_mod, _ctx, _context_opts),
+    do: {:ok, :noop}
 
   # sobelow_skip ["DOS.BinToAtom"]
-  defp create_attachment(record, attachment_def, blob) do
+  defp create_attachment(record, attachment_def, blob, context_opts) do
     resource = record.__struct__
     attachment_resource = Info.storage_attachment_resource!(resource)
 
@@ -367,11 +387,11 @@ defmodule AshStorage.Changes.Attach do
         }
       end
 
-    Ash.create(attachment_resource, params, action: :create)
+    Ash.create(attachment_resource, params, Keyword.merge(context_opts, action: :create))
   end
 
   # sobelow_skip ["DOS.BinToAtom"]
-  defp find_attachments(record, attachment_def) do
+  defp find_attachments(record, attachment_def, context_opts) do
     resource = record.__struct__
     attachment_resource = Info.storage_attachment_resource!(resource)
     record_id = Map.get(record, :id) |> to_string()
@@ -398,16 +418,19 @@ defmodule AshStorage.Changes.Attach do
     attachment_resource
     |> Ash.Query.filter(^filter)
     |> Ash.Query.load(:blob)
+    |> Ash.Query.set_tenant(context_opts[:tenant])
     |> Ash.read()
   end
 
-  defp purge_attachments(attachments, service_mod, ctx) do
+  defp purge_attachments(attachments, service_mod, ctx, context_opts) do
+    destroy_opts = Keyword.merge(context_opts, action: :destroy, return_destroyed?: true)
+
     Enum.reduce_while(attachments, {:ok, []}, fn att, {:ok, acc} ->
       blob = att.blob
 
       with :ok <- service_mod.delete(blob.key, ctx),
-           {:ok, _} <- Ash.destroy(att, action: :destroy, return_destroyed?: true),
-           {:ok, _} <- Ash.destroy(blob, action: :destroy, return_destroyed?: true) do
+           {:ok, _} <- Ash.destroy(att, destroy_opts),
+           {:ok, _} <- Ash.destroy(blob, destroy_opts) do
         {:cont, {:ok, [att | acc]}}
       else
         {:error, error} -> {:halt, {:error, error}}

lib/ash_storage/changes/attach_blob.ex

@@ -42,7 +42,7 @@ defmodule AshStorage.Changes.AttachBlob do
 
   # sobelow_skip ["DOS.BinToAtom"]
   @impl true
-  def change(changeset, opts, _context) do
+  def change(changeset, opts, context) do
     argument_name = opts[:argument]
     attachment_name = opts[:attachment]
 
@@ -53,13 +53,16 @@ defmodule AshStorage.Changes.AttachBlob do
         {:ok, record}
       else
         resource = record.__struct__
+        context_opts = Ash.Context.to_opts(context)
 
         with {:ok, attachment_def} <- Info.attachment(resource, attachment_name),
              {:ok, {service_mod, service_opts}} <- resolve_service(resource, attachment_def),
              ctx = build_context(service_opts, resource, attachment_def, changeset),
-             {:ok, blob} <- fetch_blob(resource, blob_id),
-             {:ok, _} <- maybe_replace_existing(record, attachment_def, service_mod, ctx),
-             {:ok, attachment} <- create_attachment(record, attachment_def, blob) do
+             {:ok, blob} <- fetch_blob(resource, blob_id, context_opts),
+             {:ok, _} <-
+               maybe_replace_existing(record, attachment_def, service_mod, ctx, context_opts),
+             {:ok, attachment} <-
+               create_attachment(record, attachment_def, blob, context_opts) do
           record =
             record
             |> Ash.Resource.put_metadata(:"#{attachment_name}_blob", blob)
@@ -87,26 +90,33 @@ defmodule AshStorage.Changes.AttachBlob do
     )
   end
 
-  defp fetch_blob(resource, blob_id) do
+  defp fetch_blob(resource, blob_id, context_opts) do
     blob_resource = Info.storage_blob_resource!(resource)
 
-    case Ash.get(blob_resource, blob_id) do
+    case Ash.get(blob_resource, blob_id, context_opts) do
       {:ok, blob} -> {:ok, blob}
       {:error, _} -> {:error, :blob_not_found}
     end
   end
 
-  defp maybe_replace_existing(record, %{type: :one} = attachment_def, service_mod, ctx) do
-    case find_attachments(record, attachment_def) do
+  defp maybe_replace_existing(
+         record,
+         %{type: :one} = attachment_def,
+         service_mod,
+         ctx,
+         context_opts
+       ) do
+    case find_attachments(record, attachment_def, context_opts) do
       {:ok, []} -> {:ok, :noop}
-      {:ok, existing} -> purge_attachments(existing, service_mod, ctx)
+      {:ok, existing} -> purge_attachments(existing, service_mod, ctx, context_opts)
     end
   end
 
-  defp maybe_replace_existing(_record, %{type: :many}, _service_mod, _ctx), do: {:ok, :noop}
+  defp maybe_replace_existing(_record, %{type: :many}, _service_mod, _ctx, _context_opts),
+    do: {:ok, :noop}
 
   # sobelow_skip ["DOS.BinToAtom"]
-  defp create_attachment(record, attachment_def, blob) do
+  defp create_attachment(record, attachment_def, blob, context_opts) do
     resource = record.__struct__
     attachment_resource = Info.storage_attachment_resource!(resource)
     record_id = Map.get(record, :id) |> to_string()
@@ -137,11 +147,11 @@ defmodule AshStorage.Changes.AttachBlob do
         }
       end
 
-    Ash.create(attachment_resource, params, action: :create)
+    Ash.create(attachment_resource, params, Keyword.merge(context_opts, action: :create))
   end
 
   # sobelow_skip ["DOS.BinToAtom"]
-  defp find_attachments(record, attachment_def) do
+  defp find_attachments(record, attachment_def, context_opts) do
     resource = record.__struct__
     attachment_resource = Info.storage_attachment_resource!(resource)
     record_id = Map.get(record, :id) |> to_string()
@@ -168,16 +178,19 @@ defmodule AshStorage.Changes.AttachBlob do
     attachment_resource
     |> Ash.Query.filter(^filter)
     |> Ash.Query.load(:blob)
+    |> Ash.Query.set_tenant(context_opts[:tenant])
     |> Ash.read()
   end
 
-  defp purge_attachments(attachments, service_mod, ctx) do
+  defp purge_attachments(attachments, service_mod, ctx, context_opts) do
+    destroy_opts = Keyword.merge(context_opts, action: :destroy, return_destroyed?: true)
+
     Enum.reduce_while(attachments, {:ok, []}, fn att, {:ok, acc} ->
       blob = att.blob
 
       with :ok <- service_mod.delete(blob.key, ctx),
-           {:ok, _} <- Ash.destroy(att, action: :destroy, return_destroyed?: true),
-           {:ok, _} <- Ash.destroy(blob, action: :destroy, return_destroyed?: true) do
+           {:ok, _} <- Ash.destroy(att, destroy_opts),
+           {:ok, _} <- Ash.destroy(blob, destroy_opts) do
         {:cont, {:ok, [att | acc]}}
       else
         {:error, error} -> {:halt, {:error, error}}

lib/ash_storage/changes/handle_file_argument.ex

@@ -294,11 +294,10 @@ defmodule AshStorage.Changes.HandleFileArgument do
     checksum = :crypto.hash(:md5, data) |> Base.encode64()
     byte_size = byte_size(data)
 
-    with :ok <- service_mod.upload(key, data, ctx) do
+    with {:ok, extra_blob_attrs} <- normalize_upload(service_mod.upload(key, data, ctx)) do
       blob_resource = Info.storage_blob_resource!(resource)
 
-      Ash.create(
-        blob_resource,
+      blob_attrs =
         %{
           key: key,
           filename: filename,
@@ -308,12 +307,17 @@ defmodule AshStorage.Changes.HandleFileArgument do
           service_name: service_mod,
           service_opts: persistable_service_opts(service_mod, ctx.service_opts),
           metadata: %{}
-        },
-        action: :create
-      )
+        }
+        |> Map.merge(extra_blob_attrs)
+
+      Ash.create(blob_resource, blob_attrs, action: :create)
     end
   end
 
+  defp normalize_upload(:ok), do: {:ok, %{}}
+  defp normalize_upload({:ok, attrs}) when is_map(attrs), do: {:ok, attrs}
+  defp normalize_upload({:error, _} = error), do: error
+
   defp read_io(%Ash.Type.File{} = file) do
     {:ok, device} = Ash.Type.File.open(file, [:read, :binary])
     data = IO.binread(device, :eof)

lib/ash_storage/operations.ex

@@ -72,11 +72,14 @@ defmodule AshStorage.Operations do
          {:ok, {service_mod, service_opts}} <- resolve_service(resource, attachment_def) do
       ctx = build_context(service_opts, resource, attachment_def, opts)
 
-      filename = Keyword.fetch!(opts, :filename)
-      content_type = Keyword.get(opts, :content_type, "application/octet-stream")
-      byte_size = Keyword.get(opts, :byte_size, 0)
-      checksum = Keyword.get(opts, :checksum, "")
-      metadata = Keyword.get(opts, :metadata, %{})
+      {arg_opts, action_opts} =
+        Keyword.split(opts, [:filename, :content_type, :byte_size, :checksum, :metadata])
+
+      filename = Keyword.fetch!(arg_opts, :filename)
+      content_type = Keyword.get(arg_opts, :content_type, "application/octet-stream")
+      byte_size = Keyword.get(arg_opts, :byte_size, 0)
+      checksum = Keyword.get(arg_opts, :checksum, "")
+      metadata = Keyword.get(arg_opts, :metadata, %{})
 
       key = AshStorage.generate_key()
       blob_resource = Info.storage_blob_resource!(resource)
@@ -94,7 +97,7 @@ defmodule AshStorage.Operations do
                  service_opts: persistable_service_opts(service_mod, service_opts),
                  metadata: metadata
                },
-               action: :create
+               Keyword.merge(action_opts, action: :create)
              ),
            {:ok, upload_info} <- service_mod.direct_upload(key, ctx) do
         {:ok, Map.put(upload_info, :blob, blob)}

lib/ash_storage/service.ex

@@ -32,9 +32,13 @@ defmodule AshStorage.Service do
 
   @doc """
   Upload a file to the storage service.
+
+  May return `:ok` or `{:ok, extra_blob_attrs}`. When a map is returned, its entries
+  are merged into the blob record on creation. This allows wrapping services (e.g.
+  encryption) to store per-file metadata such as encryption keys on the blob.
   """
   @callback upload(key(), iodata() | File.Stream.t(), Context.t()) ::
-              :ok | {:error, term()}
+              :ok | {:ok, map()} | {:error, term()}
 
   @doc """
   Download a file from the storage service.