How to purge rows from tables to s3 from a Elixir application using Ports - audit_log_purge.ex

audit_log_purge.ex

defmodule Mix.Tasks.AuditLogPurge do
  use Mix.Task
  alias MyApp.Repo
  alias MyApp.AuditLog
  def run(argv, backup_cmd \\ &build_backup_cmd/3) do
    Mix.EctoSQL.ensure_started(Repo, [])
    {options, _argv} =
      OptionParser.parse!(
        argv,
        strict: [
          month: :integer,
          year: :integer
        ]
      )
    case options do
      [year: year, month: month] ->
        purge_audit_logs(year, month)
      _ ->
        Mix.raise("""
        Lacking required year and month arguments.
        Type `mix help audit_log_purge` to know more details.
        """)
      end
    end
  end
  def purge_audit_logs(year, month) do
    audit_logs =
      from(h in AuditLog)
      |> where([i], fragment("date_part('year', ?)", i.logged_at) == ^year)
      |> where([i], fragment("date_part('month', ?)", i.logged_at) == ^month)
    port = Port.open({:spawn, backup_cmd.(year, month)}, [:binary])
    try do
      execute_cmd = fn _repo, changes ->
        income_accesses
        |> Repo.stream(log: false)
        |> Enum.each(fn row ->
          encoded_row =
            row
            |> Jason.encode!(pretty: true)
          Port.command(port, encoded_row)
          Port.command(port, "\n")
        end)
        {:ok, changes}
      end
      Multi.new()
        |> Multi.run(:execute_cmd, execute_cmd)
        |> Multi.delete_all(:audit_logs, audit_logs)
        |> Repo.transaction(log: false)
      Mix.shell().info("Audit Logs purged successfully!")
    rescue
      exception ->
        Mix.shell().error("Audit Logs failed! Message: #{exception.message}")
        reraise exception, __STACKTRACE__
    after
      Port.close(port)
    end
  end
  def build_backup_cmd(year, month) do
    "gzip -9 | aws s3 cp - s3://rk-one-api-auditlogs/#{year}/#{month}/audit_logs.json.gz --region=us-west-2"
  end
end