flow

defmodule Flow.Window.Periodic do
  @moduledoc false

  @enforce_keys [:duration]
  defstruct [:duration, :trigger, periodically: []]

  def materialize(%{duration: duration}, reducer_acc, reducer_fun, reducer_trigger, _options) do
    ref = make_ref()

    acc = fn ->
      timer = send_after(ref, duration)
      {0, timer, reducer_acc.()}
    end

    fun =
      if is_function(reducer_fun, 4) do
        fn ref, events, {window, timer, acc}, index ->
          {emit, acc} = reducer_fun.(ref, events, acc, index)
          {emit, {window, timer, acc}}
        end
      else
        fn ref, events, {window, timer, acc}, index ->
          {emit, acc} = reducer_fun.(ref, events, acc, index, {:periodic, window, :placeholder})
          {emit, {window, timer, acc}}
        end
      end

    trigger = fn
      {window, _timer, acc}, index, ^ref ->
        {emit, _} = reducer_trigger.(acc, index, {:periodic, window, :done})
        timer = send_after(ref, duration)
        {emit, {window + 1, timer, reducer_acc.()}}

      {window, timer, acc}, index, name ->
        if name == :done, do: cancel_after(ref, timer)
        {emit, acc} = reducer_trigger.(acc, index, {:periodic, window, name})
        {emit, {window, timer, acc}}
    end

    {acc, fun, trigger}
  end

  defp send_after(ref, duration) do
    Process.send_after(self(), {:trigger, ref}, duration)
  end

  defp cancel_after(ref, timer) do
    case Process.cancel_timer(timer) do
      false ->
        receive do
          {:trigger, ^ref} -> :ok
        after
          0 -> :ok
        end

      _ ->
        :ok
    end
  end
end