litestream

defmodule Litestream do
  @moduledoc """
  This GenServer module allows you to run [Litestream](https://litestream.io/) via a port in the background
  so that you can easily backup your SQLite database to an object store, a seperate local file, SFTP, etc.
  """

  use GenServer,
    restart: :transient,
    shutdown: 15_000

  require Logger

  alias Litestream.Downloader
  alias Litestream.Replicator

  @call_timeout 10_000

  # +--------------------------------------------------------------------+
  # |                 GenServer Public API Functions                     |
  # +--------------------------------------------------------------------+

  @doc """
  The `start_link/1` function is used to start the `Litestream` GenServer. After starting the GenServer, the process
  will download the Litestream binary and start it up to begin database replication. The `Litestream` GenServer
  expects a Keyword list with the following options:

  * `:repo` - The Ecto Repo that manages the SQLite database. REQUIRED
  * `:strategy` - The Litestream backup strategy you want to use. REQUIRED
  * `:name` - The name of the GenServer process. By default it is `Litestream`. OPTIONAL
  * `:bin_path` - If you already have access to the Litestream binary, provide the path via this
                  option so that you can skip the download step. OPTIONAL
  * `:version` - The version of Litestream that you want to download. OPTIONAL
  """
  def start_link(opts) do
    repo = Keyword.fetch!(opts, :repo)
    repo_config = repo.config()
    database_file = Keyword.fetch!(repo_config, :database)

    strategy =
      opts
      |> Keyword.fetch!(:strategy)
      |> maybe_create_temp_file(database_file)

    state = %{
      repo: repo,
      strategy: strategy,
      bin_path: Keyword.get(opts, :bin_path, :download),
      version: Keyword.get(opts, :version, Downloader.default_version())
    }

    GenServer.start_link(__MODULE__, state, name: Keyword.get(opts, :name, __MODULE__))
  end

  @doc """
  This function will return the status of the Litestream port with either a `:down` or
  `:running` atom.
  """
  def status(name \\ __MODULE__) do
    GenServer.call(name, :status, @call_timeout)
  end

  @doc """
  This function will cleanly stop the Litestream process, but the GenServer will still be
  running.
  """
  def stop_litestream(name \\ __MODULE__) do
    GenServer.call(name, :stop_litestream, @call_timeout)
  end

  @doc """
  This function will start the Litestream process, if it is not currently running. If it is
  already running, then this operation is effectively a no-op.
  """
  def start_litestream(name \\ __MODULE__) do
    GenServer.call(name, :start_litestream, @call_timeout)
  end

  # +------------------------------------------------------------------+
  # |                 GenServer Callback Functions                     |
  # +------------------------------------------------------------------+

  @impl true
  def init(state) do
    repo_config = state.repo.config()
    otp_app = Keyword.fetch!(repo_config, :otp_app)
    database_file = Keyword.fetch!(repo_config, :database)

    # Make sure that the process traps exits so that we can cleanly shutdown the
    # Litestream replication process
    Process.flag(:trap_exit, true)

    updated_state =
      state
      |> Map.put(:otp_app, otp_app)
      |> Map.put(:database, database_file)
      |> clear_pids()

    if state.bin_path == :download do
      {:ok, updated_state, {:continue, :download_litestream}}
    else
      unless File.exists?(state.bin_path) do
        raise "The path to the Litestream binary does not exist: #{inspect(state.bin_path)}"
      end

      {:ok, updated_state, {:continue, :start_litestream}}
    end
  end

  @impl true
  def handle_continue(:download_litestream, %{otp_app: otp_app, version: version} = state) do
    otp_app_priv_dir = :code.priv_dir(otp_app)
    download_dir = Path.join(otp_app_priv_dir, "/litestream/download")
    bin_dir = Path.join(otp_app_priv_dir, "/litestream/bin")

    File.mkdir_p!(download_dir)
    File.mkdir_p!(bin_dir)

    bin_path =
      case Downloader.download(bin_dir, override_version: version) do
        {:ok, output_files, []} ->
          Enum.find(output_files, fn file ->
            String.ends_with?(file, "litestream")
          end)

        {:skip, bin_path} ->
          bin_path
      end

    updated_state = Map.put(state, :bin_path, bin_path)

    {:noreply, updated_state, {:continue, :start_litestream}}
  end

  def handle_continue(:start_litestream, state) do
    {:ok, port_pid, os_pid} =
      [
        state.bin_path,
        "replicate"
        | Replicator.cli_args(state.strategy, state.database)
      ]
      |> Enum.join(" ")
      |> :exec.run_link([
        :monitor,
        {:env,
         [
           :clear | Replicator.env_vars(state.strategy)
         ]},
        {:kill_timeout, 10},
        :stdout,
        :stderr
      ])

    updated_state =
      state
      |> Map.put(:port_pid, port_pid)
      |> Map.put(:os_pid, os_pid)

    {:noreply, updated_state}
  end

  @impl true
  def handle_call(:status, _from, %{os_pid: os_pid} = state) do
    if os_pid in :exec.which_children() do
      {:reply, :running, state}
    else
      {:reply, :down, state}
    end
  end

  def handle_call(:start_litestream, _from, %{os_pid: os_pid} = state) do
    if os_pid in :exec.which_children() do
      Logger.info("Litestream is already running")

      {:reply, :ok, state}
    else
      Logger.info("Starting Litestream")

      {:reply, :ok, state, {:continue, :start_litestream}}
    end
  end

  def handle_call(:stop_litestream, _from, %{port_pid: port_pid, os_pid: os_pid} = state) do
    if os_pid in :exec.which_children() do
      :ok = :exec.kill(port_pid, :sigterm)

      {:reply, :ok, clear_pids(state)}
    else
      Logger.info("Litestream is not running")

      {:reply, :ok, state}
    end
  end

  @impl true
  def handle_info({:EXIT, _os_pid, reason}, state) do
    Logger.info("Litestream has exited with reason: #{inspect(reason)}")

    {:noreply, clear_pids(state)}
  end

  def handle_info({:DOWN, _os_pid, _process, _pid, reason}, state) do
    Logger.info("Litestream has shutdown with reason: #{reason}")

    {:noreply, state}
  end

  def handle_info({:stdout, _os_pid, output}, state) do
    Logger.info(output)

    {:noreply, state}
  end

  def handle_info({:stderr, _os_pid, output}, state) do
    Logger.warning(output)

    {:noreply, state}
  end

  @impl true
  def terminate(reason, state) do
    Logger.info("Litestream is terminating with reason #{inspect(reason)}")

    # Clean up any temp files created by the strategies
    case state.strategy do
      %_{temp_config_path: config_path} ->
        File.rm(config_path)

      _ ->
        :no_op
    end

    :ok
  end

  # +------------------------------------------------------------------+
  # |                   Private Helper Functions                       |
  # +------------------------------------------------------------------+
  defp clear_pids(state) do
    state
    |> Map.put(:port_pid, nil)
    |> Map.put(:os_pid, nil)
  end

  defp maybe_create_temp_file(strategy, database) do
    case Replicator.temp_file_contents(strategy, database) do
      nil ->
        strategy

      file_contents ->
        temp_path = Path.join(System.tmp_dir!(), "litestream-#{:erlang.unique_integer([:positive])}.yml")
        File.write!(temp_path, file_contents)
        Map.put(strategy, :temp_config_path, temp_path)
    end
  end
end