defmodule ExCmd.Stream do
  @moduledoc """
  Defines a `ExCmd.Stream` struct returned by `ExCmd.stream!/2`.
  """

  alias ExCmd.Process
  alias ExCmd.Process.Error

  require Logger

  defmodule AbnormalExit do
    defexception [:message, :exit_status]

    @impl true
    def exception(:epipe) do
      msg = "program exited due to :epipe error"
      %__MODULE__{message: msg, exit_status: :epipe}
    end

    def exception(exit_status) do
      msg = "program exited with exit status: #{exit_status}"
      %__MODULE__{message: msg, exit_status: exit_status}
    end
  end

  defmodule Sink do
    @moduledoc false

    @type t :: %__MODULE__{process: Process.t(), ignore_epipe: boolean}

    defstruct [:process, :ignore_epipe]

    defimpl Collectable do
      def into(%{process: process}) do
        collector_fun = fn
          :ok, {:cont, x} ->
            case Process.write(process, x) do
              {:error, :epipe} ->
                # there is no other way to stop a Collectable than to
                # raise error, we catch this error and return `{:error, :epipe}`
                raise Error, "epipe"

              :ok ->
                :ok
            end

          acc, :done ->
            acc

          acc, :halt ->
            acc
        end

        {:ok, collector_fun}
      end
    end
  end

  defstruct [:stream_opts, :process_opts, :cmd_with_args]

  @typedoc "Struct members are private, do not depend on them"
  @type t :: %__MODULE__{
          stream_opts: map(),
          process_opts: keyword(),
          cmd_with_args: [String.t()]
        }

  @stream_opts [
    :exit_timeout,
    :max_chunk_size,
    :input,
    :stderr,
    :ignore_epipe,
    :stream_exit_status
  ]

  @doc false
  @spec __build__(nonempty_list(String.t()), keyword()) :: t()
  def __build__(cmd_with_args, opts) do
    {stream_opts, process_opts} = Keyword.split(opts, @stream_opts)

    case normalize_stream_opts(stream_opts) do
      {:ok, stream_opts} ->
        %ExCmd.Stream{
          stream_opts: stream_opts,
          process_opts: process_opts,
          cmd_with_args: cmd_with_args
        }

      {:error, error} ->
        raise ArgumentError, message: error
    end
  end

  defimpl Enumerable do
    # credo:disable-for-next-line Credo.Check.Refactor.CyclomaticComplexity
    def reduce(arg, acc, fun) do
      start_fun = fn ->
        state = start_process(arg)
        {state, :running}
      end

      next_fun = fn
        {state, :exited} ->
          {:halt, {state, :exited}}

        {state, exit_state} ->
          %{
            process: process,
            stream_opts: %{
              stream_exit_status: stream_exit_status,
              max_chunk_size: max_chunk_size
            }
          } = state

          case Process.read(process, max_chunk_size) do
            :eof when stream_exit_status == false ->
              {:halt, {state, :eof}}

            :eof when stream_exit_status == true ->
              elem = [await_exit(state, :eof)]
              {elem, {state, :exited}}

            {:ok, x} ->
              elem = [IO.iodata_to_binary(x)]
              {elem, {state, exit_state}}

            {:error, errno} ->
              raise Error, "failed to read from the external process. errno: #{inspect(errno)}"
          end
      end

      after_fun = fn
        {_state, :exited} ->
          :ok

        {state, exit_state} ->
          case await_exit(state, exit_state) do
            {:exit, {:status, 0}} ->
              :ok

            {:exit, {:status, exit_status}} ->
              raise AbnormalExit, exit_status

            {:exit, :epipe} ->
              raise AbnormalExit, :epipe
          end
      end

      Stream.resource(start_fun, next_fun, after_fun).(acc, fun)
    end

    def count(_stream) do
      {:error, __MODULE__}
    end

    def member?(_stream, _term) do
      {:error, __MODULE__}
    end

    def slice(_stream) do
      {:error, __MODULE__}
    end

    defp start_process(%ExCmd.Stream{
           process_opts: process_opts,
           stream_opts: stream_opts,
           cmd_with_args: cmd_with_args
         }) do
      process_opts = Keyword.put(process_opts, :stderr, stream_opts[:stderr])
      {:ok, process} = Process.start_link(cmd_with_args, process_opts)
      sink = %Sink{process: process, ignore_epipe: stream_opts[:ignore_epipe]}
      writer_task = start_input_streamer(sink, stream_opts.input)

      %{process: process, stream_opts: stream_opts, writer_task: writer_task}
    end

    @doc false
    @spec start_input_streamer(term, term) :: Task.t()
    defp start_input_streamer(%Sink{process: process} = sink, input) do
      case input do
        :no_input ->
          # use `Task.completed(:ok)` when bumping min Elixir requirement
          Task.async(fn -> :ok end)

        {:enumerable, enum} ->
          Task.async(fn ->
            Process.change_pipe_owner(process, :stdin, self())

            try do
              Enum.into(enum, sink)
            rescue
              Error ->
                {:error, :epipe}
            end
          end)

        {:collectable, func} ->
          Task.async(fn ->
            Process.change_pipe_owner(process, :stdin, self())

            try do
              func.(sink)
            rescue
              Error ->
                {:error, :epipe}
            end
          end)
      end
    end

    defp await_exit(state, exit_state) do
      %{
        process: process,
        stream_opts: %{ignore_epipe: ignore_epipe, exit_timeout: exit_timeout},
        writer_task: writer_task
      } = state

      result = Process.await_exit(process, exit_timeout)
      writer_task_status = Task.await(writer_task)

      case {exit_state, result, writer_task_status} do
        # if reader exit early and there is a pending write
        {:running, {:ok, _status}, {:error, :epipe}} when ignore_epipe ->
          {:exit, {:status, 0}}

        # :killed might be due to SIGPIPE / EPIPE
        {:running, {:error, :killed}, {:error, :epipe}} when ignore_epipe ->
          {:exit, {:status, 0}}

        # if reader exit early and there is no pending write or if
        # there is no writer
        {:running, {:ok, _status}, :ok} when ignore_epipe ->
          {:exit, {:status, 0}}

        {:running, {:error, :killed}, :ok} when ignore_epipe ->
          {:exit, {:status, 0}}

        # if we get epipe from writer then raise that error, and ignore exit status
        {:running, {:ok, _status}, {:error, :epipe}} when ignore_epipe == false ->
          {:exit, :epipe}

        {:running, {:error, :killed}, {:error, :epipe}} when ignore_epipe == false ->
          {:exit, :epipe}

        # Normal exit success case
        {_, {:ok, 0}, _} ->
          {:exit, {:status, 0}}

        {:eof, {:ok, exit_status}, _} ->
          {:exit, {:status, exit_status}}
      end
    end
  end

  @spec normalize_input(term) ::
          {:ok, :no_input} | {:ok, {:enumerable, term}} | {:ok, {:collectable, function}}
  defp normalize_input(term) do
    cond do
      is_nil(term) ->
        {:ok, :no_input}

      !is_function(term, 1) && Enumerable.impl_for(term) ->
        {:ok, {:enumerable, term}}

      is_function(term, 1) ->
        {:ok, {:collectable, term}}

      is_binary(term) ->
        {:ok, {:enumerable, [term]}}

      true ->
        {:error,
         ":input must be a string, an Enumerable, or a function that accepts a Collectable."}
    end
  end

  defp normalize_max_chunk_size(max_chunk_size) do
    case max_chunk_size do
      nil ->
        {:ok, 65_531}

      max_chunk_size when is_integer(max_chunk_size) and max_chunk_size > 0 ->
        {:ok, max_chunk_size}

      _ ->
        {:error, ":max_chunk_size must be a positive integer"}
    end
  end

  defp normalize_exit_timeout(timeout) do
    case timeout do
      nil ->
        {:ok, 5000}

      timeout when is_integer(timeout) and timeout > 0 ->
        {:ok, timeout}

      _ ->
        {:error, ":exit_timeout must be either :infinity or an integer"}
    end
  end

  defp normalize_stderr(stderr) do
    case stderr do
      nil ->
        {:ok, :disable}

      stderr when stderr in [:console, :redirect_to_stdout, :disable] ->
        {:ok, stderr}

      _ ->
        {:error, ":stderr must be an atom and one of :console, :redirect_to_stdout, :disable"}
    end
  end

  defp normalize_ignore_epipe(ignore_epipe) do
    case ignore_epipe do
      nil ->
        {:ok, false}

      ignore_epipe when is_boolean(ignore_epipe) ->
        {:ok, ignore_epipe}

      _ ->
        {:error, ":ignore_epipe must be a boolean"}
    end
  end

  defp normalize_stream_exit_status(stream_exit_status) do
    case stream_exit_status do
      nil ->
        {:ok, false}

      stream_exit_status when is_boolean(stream_exit_status) ->
        {:ok, stream_exit_status}

      _ ->
        {:error, ":stream_exit_status must be a boolean"}
    end
  end

  defp normalize_stream_opts(opts) do
    with {:ok, input} <- normalize_input(opts[:input]),
         {:ok, exit_timeout} <- normalize_exit_timeout(opts[:exit_timeout]),
         {:ok, max_chunk_size} <- normalize_max_chunk_size(opts[:max_chunk_size]),
         {:ok, stderr} <- normalize_stderr(opts[:stderr]),
         {:ok, ignore_epipe} <- normalize_ignore_epipe(opts[:ignore_epipe]),
         {:ok, stream_exit_status} <- normalize_stream_exit_status(opts[:stream_exit_status]) do
      {:ok,
       %{
         input: input,
         exit_timeout: exit_timeout,
         max_chunk_size: max_chunk_size,
         stderr: stderr,
         ignore_epipe: ignore_epipe,
         stream_exit_status: stream_exit_status
       }}
    end
  end
end
