Upload to S3

- Upload a timestamped file
- Update a stable copy to the latest file

apps/transport/lib/S3/s3.ex

@@ -3,7 +3,7 @@ defmodule Transport.S3 do
   This module contains common code related to S3 object storage.
   """
   require Logger
-  @type bucket_feature :: :history | :on_demand_validation | :gtfs_diff | :logos
+  @type bucket_feature :: :history | :on_demand_validation | :gtfs_diff | :logos | :stops_registry
 
   @spec bucket_name(bucket_feature()) :: binary()
   def bucket_name(feature) do
@@ -55,4 +55,12 @@ defmodule Transport.S3 do
     |> ExAws.S3.download_file(remote_path, local_path)
     |> Transport.Wrapper.ExAWS.impl().request!()
   end
+
+  @spec remote_copy_file(bucket_feature(), binary(), binary()) :: any()
+  def remote_copy_file(feature, remote_path_src, remote_path_dest) do
+    bucket = Transport.S3.bucket_name(feature)
+
+    ExAws.S3.put_object_copy(bucket, remote_path_dest, bucket, remote_path_src)
+    |> Transport.Wrapper.ExAWS.impl().request!()
+  end
 end

apps/transport/lib/jobs/stops_registry_snapshot_job.ex

@@ -2,14 +2,43 @@ defmodule Transport.Jobs.StopsRegistrySnapshotJob do
   @moduledoc """
   Job in charge of building a snapshot of the stops registry.
   """
-
   use Oban.Worker, unique: [period: {1, :days}], tags: ["registry"], max_attempts: 3
   require Logger
 
   @impl Oban.Worker
-  def perform(_job) do
+  def perform(%Oban.Job{}) do
     file = "#{System.tmp_dir!()}/registre-arrets.csv"
 
-    Transport.Registry.Engine.execute(file)
+    :ok = Transport.Registry.Engine.execute(file)
+
+    file
+    |> compress()
+    |> upload("stops_registry_#{timestamp()}.csv.zip")
+    |> link("stops_registry_latest.csv.zip")
+  end
+
+  defp timestamp do
+    DateTime.utc_now()
+    |> Calendar.strftime("%Y%m%d.%H%M%S.%f")
+  end
+
+  defp compress(file) do
+    archive = "#{file}.zip"
+
+    {:ok, _filename} =
+      :zip.create(
+        archive,
+        [String.to_charlist(file)]
+      )
+
+    archive
+  end
+
+  defp upload(file, filename) do
+    Transport.S3.stream_to_s3!(:stops_registry, file, filename)
+  end
+
+  defp link(s3_path, filename) do
+    Transport.S3.remote_copy_file(:stops_registry, s3_path, filename)
   end
 end

config/dev.exs

@@ -72,7 +72,8 @@ config :transport,
     history: "resource-history-dev",
     on_demand_validation: "on-demand-validation-dev",
     gtfs_diff: "gtfs-diff-dev",
-    logos: "logos-dev"
+    logos: "logos-dev",
+    stops_registry: "stops-registry-dev"
   }
 
 config :oauth2, Datagouvfr.Authentication,

config/prod.exs

@@ -9,7 +9,8 @@ config :transport,
     history: "resource-history-prod",
     on_demand_validation: "on-demand-validation-prod",
     gtfs_diff: "gtfs-diff-prod",
-    logos: "logos-prod"
+    logos: "logos-prod",
+    stops_registry: "stops-registry-prod"
   }
 
 # Configure Sentry for production and staging.

config/runtime.exs

@@ -94,7 +94,8 @@ if app_env == :staging do
       history: "resource-history-staging",
       on_demand_validation: "on-demand-validation-staging",
       gtfs_diff: "gtfs-diff-staging",
-      logos: "logos-staging"
+      logos: "logos-staging",
+      stops_registry: "stops-registry-staging"
     }
 end
 

config/test.exs

@@ -45,7 +45,8 @@ config :transport,
     history: "resource-history-test",
     on_demand_validation: "on-demand-validation-test",
     gtfs_diff: "gtfs-diff-test",
-    logos: "logos-test"
+    logos: "logos-test",
+    stops_registry: "stops-registry-test"
   },
   workflow_notifier: Transport.Jobs.Workflow.ProcessNotifier,
   export_secret_key: "fake_export_secret_key",