Add live replication support

config/runtime.exs

@@ -31,7 +31,6 @@ config :philomena,
   tag_url_root: System.fetch_env!("TAG_URL_ROOT"),
   redis_host: System.get_env("REDIS_HOST", "localhost"),
   proxy_host: System.get_env("PROXY_HOST"),
-  s3_bucket: System.fetch_env!("S3_BUCKET"),
   camo_host: System.get_env("CAMO_HOST"),
   camo_key: System.get_env("CAMO_KEY"),
   cdn_host: System.fetch_env!("CDN_HOST")
@@ -67,11 +66,26 @@ if is_nil(System.get_env("START_WORKER")) do
   config :exq, queues: []
 end
 
-# S3 config
-config :ex_aws, :s3,
+# S3/Object store config
+config :philomena, :s3_primary_options,
+  region: System.get_env("S3_REGION", "us-east-1"),
   scheme: System.fetch_env!("S3_SCHEME"),
   host: System.fetch_env!("S3_HOST"),
-  port: System.fetch_env!("S3_PORT")
+  port: System.fetch_env!("S3_PORT"),
+  access_key_id: System.fetch_env!("AWS_ACCESS_KEY_ID"),
+  secret_access_key: System.fetch_env!("AWS_SECRET_ACCESS_KEY")
+
+config :philomena, :s3_primary_bucket, System.fetch_env!("S3_BUCKET")
+
+config :philomena, :s3_secondary_options,
+  region: System.get_env("ALT_S3_REGION", "us-east-1"),
+  scheme: System.get_env("ALT_S3_SCHEME"),
+  host: System.get_env("ALT_S3_HOST"),
+  port: System.get_env("ALT_S3_PORT"),
+  access_key_id: System.get_env("ALT_AWS_ACCESS_KEY_ID"),
+  secret_access_key: System.get_env("ALT_AWS_SECRET_ACCESS_KEY")
+
+config :philomena, :s3_secondary_bucket, System.get_env("ALT_S3_BUCKET")
 
 if config_env() != :test do
   # Database config

lib/mix/tasks/upload_to_s3.ex

@@ -10,9 +10,8 @@ defmodule Mix.Tasks.UploadToS3 do
   }
 
   alias Philomena.Images.Thumbnailer
-  alias Philomena.Mime
+  alias Philomena.Objects
   alias Philomena.Batch
-  alias ExAws.S3
   import Ecto.Query
 
   @shortdoc "Dumps existing image files to S3 storage backend"
@@ -139,14 +138,6 @@ defmodule Mix.Tasks.UploadToS3 do
   end
 
   defp put_file(path, uploaded_path) do
-    {_, mime} = Mime.file(path)
-    contents = File.read!(path)
-
-    S3.put_object(bucket(), uploaded_path, contents, content_type: mime)
-    |> ExAws.request!()
-  end
-
-  defp bucket do
-    Application.fetch_env!(:philomena, :s3_bucket)
+    Objects.put(uploaded_path, path)
   end
 end

lib/philomena/images/thumbnailer.ex

@@ -9,9 +9,9 @@ defmodule Philomena.Images.Thumbnailer do
   alias Philomena.Processors
   alias Philomena.Analyzers
   alias Philomena.Uploader
+  alias Philomena.Objects
   alias Philomena.Sha512
   alias Philomena.Repo
-  alias ExAws.S3
 
   @versions [
     thumb_tiny: {50, 50},
@@ -122,7 +122,7 @@ defmodule Philomena.Images.Thumbnailer do
     tempfile = Briefly.create!(extname: ".#{image.image_format}")
     path = Path.join(image_thumb_prefix(image), "full.#{image.image_format}")
 
-    ExAws.request!(S3.download_file(bucket(), path, tempfile))
+    Objects.download_file(path, tempfile)
 
     tempfile
   end
@@ -138,17 +138,15 @@ defmodule Philomena.Images.Thumbnailer do
       source = Path.join(source_prefix, name)
       target = Path.join(target_prefix, name)
 
-      ExAws.request(S3.put_object_copy(bucket(), target, bucket(), source))
-      ExAws.request(S3.delete_object(bucket(), source))
+      Objects.copy(source, target)
+      Objects.delete(source)
     end)
   end
 
   defp bulk_delete(file_names, prefix) do
-    Enum.map(file_names, fn name ->
-      target = Path.join(prefix, name)
-
-      ExAws.request(S3.delete_object(bucket(), target))
-    end)
+    file_names
+    |> Enum.map(&Path.join(prefix, &1))
+    |> Objects.delete_multiple()
   end
 
   def all_versions(image) do
@@ -189,7 +187,4 @@ defmodule Philomena.Images.Thumbnailer do
 
   defp image_url_root,
     do: Application.fetch_env!(:philomena, :image_url_root)
-
-  defp bucket,
-    do: Application.fetch_env!(:philomena, :s3_bucket)
 end

lib/philomena/objects.ex

@@ -0,0 +1,104 @@
+defmodule Philomena.Objects do
+  @moduledoc """
+  Replication wrapper for object storage backends.
+  """
+  alias Philomena.Mime
+
+  #
+  # Fetch a key from the primary storage backend and
+  # write it into the destination file.
+  #
+  @spec download_file(String.t(), String.t()) :: any()
+  def download_file(key, file_path) do
+    [opts] = primary_opts()
+
+    contents =
+      ExAws.S3.get_object(opts[:bucket], key)
+      |> ExAws.request!(opts[:config_overrides])
+
+    File.write!(file_path, contents.body)
+  end
+
+  #
+  # Upload a file using a single API call, writing the
+  # contents from the given path to storage.
+  #
+  @spec put(String.t(), String.t()) :: any()
+  def put(key, file_path) do
+    {_, mime} = Mime.file(file_path)
+    contents = File.read!(file_path)
+
+    run_all(fn opts ->
+      ExAws.S3.put_object(opts[:bucket], key, contents, content_type: mime)
+      |> ExAws.request!(opts[:config_overrides])
+    end)
+  end
+
+  #
+  # Copies a key from the source to the destination,
+  # overwriting the destination object if its exists.
+  #
+  @spec copy(String.t(), String.t()) :: any()
+  def copy(source_key, dest_key) do
+    run_all(fn opts ->
+      ExAws.S3.put_object_copy(opts[:bucket], dest_key, opts[:bucket], source_key)
+      |> ExAws.request!(opts[:config_overrides])
+    end)
+  end
+
+  #
+  # Removes the key from storage.
+  #
+  @spec delete(String.t()) :: any()
+  def delete(key) do
+    run_all(fn opts ->
+      ExAws.S3.delete_object(opts[:bucket], key)
+      |> ExAws.request!(opts[:config_overrides])
+    end)
+  end
+
+  #
+  # Removes all given keys from storage.
+  #
+  @spec delete_multiple([String.t()]) :: any()
+  def delete_multiple(keys) do
+    run_all(fn opts ->
+      ExAws.S3.delete_multiple_objects(opts[:bucket], keys)
+      |> ExAws.request!(opts[:config_overrides])
+    end)
+  end
+
+  defp run_all(fun) do
+    backends()
+    |> Task.async_stream(fun)
+    |> Stream.run()
+  end
+
+  defp backends do
+    primary_opts() ++ replica_opts()
+  end
+
+  defp primary_opts do
+    [
+      %{
+        config_overrides: Application.fetch_env!(:philomena, :s3_primary_options),
+        bucket: Application.fetch_env!(:philomena, :s3_primary_bucket)
+      }
+    ]
+  end
+
+  defp replica_opts do
+    replica_bucket = Application.get_env(:philomena, :s3_secondary_bucket)
+
+    if not is_nil(replica_bucket) do
+      [
+        %{
+          config_overrides: Application.fetch_env!(:philomena, :s3_secondary_options),
+          bucket: replica_bucket
+        }
+      ]
+    else
+      []
+    end
+  end
+end

lib/philomena/uploader.ex

@@ -5,9 +5,8 @@ defmodule Philomena.Uploader do
 
   alias Philomena.Filename
   alias Philomena.Analyzers
+  alias Philomena.Objects
   alias Philomena.Sha512
-  alias Philomena.Mime
-  alias ExAws.S3
   import Ecto.Changeset
 
   @doc """
@@ -73,12 +72,7 @@ defmodule Philomena.Uploader do
   content type and permissions.
   """
   def persist_file(path, file) do
-    {_, mime} = Mime.file(file)
-
-    file
-    |> S3.Upload.stream_file()
-    |> S3.upload(bucket(), path, content_type: mime)
-    |> ExAws.request!()
+    Objects.put(path, file)
   end
 
   @doc """
@@ -117,9 +111,7 @@ defmodule Philomena.Uploader do
   defp try_remove(nil, _file_root), do: nil
 
   defp try_remove(file, file_root) do
-    path = Path.join(file_root, file)
-
-    ExAws.request!(S3.delete_object(bucket(), path))
+    Objects.delete(Path.join(file_root, file))
   end
 
   defp prefix_attributes(map, prefix),
@@ -130,8 +122,4 @@ defmodule Philomena.Uploader do
   defp remove_key(field_name), do: "removed_#{field_name}"
 
   defp field(field_name), do: String.to_existing_atom(field_name)
-
-  defp bucket do
-    Application.fetch_env!(:philomena, :s3_bucket)
-  end
 end