file_stream

defmodule FileStream.Storage.SFTP do
  @moduledoc """
  A module to stream SFTP backed files.
  """

  @behaviour FileStream.Storage

  @opts_schema KeywordValidator.schema!(
                 user: [is: :binary],
                 password: [is: :binary],
                 user_dir: [is: :binary],
                 inet: [is: {:one_of, [:inet, :inet6]}],
                 sftp_vsn: [is: :integer],
                 connect_timeout: [is: :timeout, required: true, default: 5_000],
                 operation_timeout: [is: :timeout, required: true, default: :infinity],
                 dsa_pass_phrase: [is: :any],
                 rsa_pass_phrase: [is: :any],
                 ecdsa_pass_phrase: [is: :any],
                 key_cb: [is: :any],
                 modify_algorithms: [is: {:one_of, [:tuple, :list]}]
               )

  ################################
  # FileStream.Storage Callbacks
  ################################

  @impl FileStream.Storage
  def init(opts) do
    opts = KeywordValidator.validate!(opts, @opts_schema)
    Enum.into(opts, %{})
  end

  @impl FileStream.Storage
  def into(stream) do
    {init_write(stream), &write_stream/2}
  end

  @impl FileStream.Storage
  def reduce(stream, acc, fun) do
    Stream.resource(
      fn -> init_read(stream) end,
      &read_stream/1,
      &exit_stream/1
    ).(acc, fun)
  end

  ################################
  # Private API
  ################################

  defp init_write(stream) do
    stream = connect(stream)
    open_file(stream, [:write, :creat, :binary])
  end

  defp init_read(stream) do
    stream = connect(stream)
    open_file(stream, [:read, :binary])
  end

  defp connect(stream) do
    opts = connect_opts(stream)

    case :ssh_sftp.start_channel(to_charlist(stream.uri.host), stream.uri.port, opts) do
      {:ok, chan_pid, conn_ref} ->
        stream
        |> FileStream.put_meta(:chan_pid, chan_pid)
        |> FileStream.put_meta(:conn_ref, conn_ref)

      {:error, 'Unable to connect using the available authentication methods'} ->
        raise FileStream.PermissionError, uri: stream.uri

      {:error, :timeout} ->
        raise FileStream.ConnectionTimeoutError, uri: stream.uri

      {:error, :nxdomain} ->
        raise FileStream.NXDomainError, uri: stream.uri

      {:error, reason} ->
        raise FileStream.Error, uri: stream.uri, reason: reason
    end
  end

  defp connect_opts(stream) do
    user = stream.uri.userinfo || stream.config.user

    stream.config
    |> Enum.into([])
    |> Keyword.put(:user, user)
    |> Keyword.take([
      :user,
      :password,
      :user_dir,
      :system_dir,
      :inet,
      :sftp_vsn,
      :connect_timeout,
      :dsa_pass_phrase,
      :rsa_pass_phrase,
      :ecdsa_pass_phrase,
      :key_cb,
      :modify_algorithms
    ])
    |> Keyword.merge(
      quiet_mode: true,
      silently_accept_hosts: true,
      user_interaction: false
    )
    |> Enum.map(fn
      {key, val} when is_binary(val) -> {key, to_charlist(val)}
      val -> val
    end)
    |> Enum.filter(fn
      {_, nil} -> false
      _ -> true
    end)
  end

  defp open_file(stream, modes) do
    chan_pid = FileStream.get_meta(stream, :chan_pid)
    path = URI.decode(stream.uri.path || "/")
    path = String.to_charlist(path)

    case :ssh_sftp.open(chan_pid, path, modes, stream.config.operation_timeout) do
      {:ok, handle} ->
        FileStream.put_meta(stream, :handle, handle)

      {:error, :no_such_file} ->
        raise FileStream.NotFoundError, uri: stream.uri

      {:error, :no_such_path} ->
        raise FileStream.NotFoundError, uri: stream.uri

      {:error, :permission_denied} ->
        raise FileStream.PermissionError, uri: stream.uri

      {:error, reason} ->
        raise FileStream.Error, uri: stream.uri, reason: reason
    end
  end

  defp write_stream(stream, command) do
    chan_pid = FileStream.get_meta(stream, :chan_pid)
    handle = FileStream.get_meta(stream, :handle)

    case command do
      {:cont, chunk} ->
        write_chunk(stream, chan_pid, handle, chunk)

      :done ->
        exit_stream(stream)

      :halt ->
        :ok
    end
  end

  defp write_chunk(stream, chan_pid, handle, chunk) do
    case :ssh_sftp.write(chan_pid, handle, chunk, stream.config.connect_timeout) do
      :ok ->
        stream

      {:error, :permission_denied} ->
        raise FileStream.PermissionError, uri: stream.uri

      {:error, reason} ->
        raise FileStream.Error, uri: stream.uri, reason: reason
    end
  end

  defp read_stream(stream) do
    chan_pid = FileStream.get_meta(stream, :chan_pid)
    handle = FileStream.get_meta(stream, :handle)

    case :ssh_sftp.read(chan_pid, handle, 32_768, stream.config.operation_timeout) do
      :eof ->
        {:halt, stream}

      {:ok, chunk} ->
        {[chunk], stream}

      {:error, :no_such_file} ->
        raise FileStream.NotFoundError, uri: stream.uri

      {:error, :no_such_path} ->
        raise FileStream.NotFoundError, uri: stream.uri

      {:error, :permission_denied} ->
        raise FileStream.PermissionError, uri: stream.uri

      {:error, reason} ->
        raise FileStream.Error, uri: stream.uri, reason: reason
    end
  end

  defp exit_stream(stream) do
    chan_pid = FileStream.get_meta(stream, :chan_pid)
    conn_ref = FileStream.get_meta(stream, :conn_ref)
    handle = FileStream.get_meta(stream, :handle)

    case :ssh_sftp.close(chan_pid, handle, stream.config.operation_timeout) do
      :ok ->
        :ok = :ssh_sftp.stop_channel(chan_pid)
        :ok = :ssh.close(conn_ref)
        FileStream.reset_meta(stream)

      {:error, reason} ->
        raise FileStream.Error, uri: stream.uri, reason: reason
    end
  end
end