file_stream

defmodule FileStream.Storage.Local do
  @moduledoc """
  A module to stream local files.
  """

  @behaviour FileStream.Storage

  @read_ahead_size 64 * 1024
  @opts_schema KeywordValidator.schema!(
                 line_or_bytes: [
                   is: {:one_of, [{:=, :line}, :integer]},
                   required: true,
                   default: :line
                 ],
                 modes: [
                   is: :list,
                   required: true,
                   default: []
                 ]
               )

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

  @impl FileStream.Storage
  def init(opts) do
    opts = KeywordValidator.validate!(opts, @opts_schema)
    opts = do_init(opts)
    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 do_init(opts) do
    modes = normalize_modes(opts[:modes], true)
    raw = :lists.keyfind(:encoding, 1, modes) == false

    modes =
      case raw do
        true ->
          case :lists.keyfind(:read_ahead, 1, modes) do
            {:read_ahead, false} -> [:raw | :lists.keydelete(:read_ahead, 1, modes)]
            {:read_ahead, _} -> [:raw | modes]
            false -> [:raw, :read_ahead | modes]
          end

        false ->
          modes
      end

    opts = Keyword.replace(opts, :modes, modes)
    Keyword.put(opts, :raw, raw)
  end

  defp init_write(stream) do
    path = build_path(stream)
    modes = for mode <- stream.config.modes, mode not in [:read], do: mode

    case :file.open(path, [:write | modes]) do
      {:ok, device} ->
        FileStream.put_meta(stream, :device, device)

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

      {:error, reason} ->
        reason = :file.format_error(reason)
        raise FileStream.Error, uri: stream.uri, reason: reason
    end
  end

  defp write_stream(stream, command) do
    case command do
      {:cont, chunk} ->
        write_chunk(stream, chunk)

      :done ->
        exit_stream(stream)

      :halt ->
        :ok
    end
  end

  defp write_chunk(stream, chunk) do
    device = FileStream.get_meta(stream, :device)

    case stream.config.raw do
      true -> :ok = IO.binwrite(device, chunk)
      false -> :ok = IO.write(device, chunk)
    end

    stream
  end

  defp exit_stream(stream) do
    device = FileStream.get_meta(stream, :device)
    :ok = :file.close(device)
    FileStream.reset_meta(stream)
  end

  defp init_read(stream) do
    path = build_path(stream)
    modes = read_modes(stream.config.modes)

    case :file.open(path, modes) do
      {:ok, device} ->
        device =
          if :trim_bom in stream.config.modes do
            {device, _} = trim_bom(device, stream.config.raw)
            device
          else
            device
          end

        FileStream.put_meta(stream, :device, device)

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

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

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

      {:error, reason} ->
        reason = :file.format_error(reason)
        raise FileStream.Error, uri: stream.uri, reason: reason
    end
  end

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

    {chunk, device} =
      case stream.config.raw do
        true -> IO.each_binstream(device, stream.config.line_or_bytes)
        false -> IO.each_stream(device, stream.config.line_or_bytes)
      end

    {chunk, FileStream.put_meta(stream, :device, device)}
  end

  defp build_path(stream) do
    "/" <> (stream.uri.host || "") <> (stream.uri.path || "")
  end

  defp read_modes(modes) do
    for mode <- modes, mode not in [:write, :append, :trim_bom], do: mode
  end

  defp trim_bom(device, true) do
    bom_length = device |> IO.binread(4) |> bom_length()
    {:ok, new_pos} = :file.position(device, bom_length)
    {device, new_pos}
  end

  defp trim_bom(device, false) do
    # Or we read the bom in the correct amount or it isn't there
    case bom_length(IO.read(device, 1)) do
      0 ->
        {:ok, _} = :file.position(device, 0)
        {device, 0}

      _ ->
        {device, 1}
    end
  end

  defp bom_length(<<239, 187, 191, _rest::binary>>), do: 3
  defp bom_length(<<254, 255, _rest::binary>>), do: 2
  defp bom_length(<<255, 254, _rest::binary>>), do: 2
  defp bom_length(<<0, 0, 254, 255, _rest::binary>>), do: 4
  defp bom_length(<<254, 255, 0, 0, _rest::binary>>), do: 4
  defp bom_length(_binary), do: 0

  defp normalize_modes([:utf8 | rest], binary?) do
    [encoding: :utf8] ++ normalize_modes(rest, binary?)
  end

  defp normalize_modes([:read_ahead | rest], binary?) do
    [read_ahead: @read_ahead_size] ++ normalize_modes(rest, binary?)
  end

  defp normalize_modes([mode | rest], _binary?) when mode in [:charlist, :char_list] do
    if mode == :char_list do
      IO.warn("the :char_list mode is deprecated, use :charlist")
    end

    normalize_modes(rest, false)
  end

  defp normalize_modes([mode | rest], binary?) do
    [mode | normalize_modes(rest, binary?)]
  end

  defp normalize_modes([], true), do: [:binary]
  defp normalize_modes([], false), do: []
end