Como processar consultas de CPF em paralelo usando GenServer e Task

defmodule CpfProcessor do
    use GenServer

    defstruct [:api_key, pendentes: [], resultados: [], em_processamento: 0, max_concurrent: 5]

    # API pública
    def start_link(opts) do
    api_key = Keyword.fetch!(opts, :api_key)
    max_concurrent = Keyword.get(opts, :max_concurrent, 5)
    GenServer.start_link(__MODULE__, %__MODULE__{api_key: api_key, max_concurrent: max_concurrent}, name: __MODULE__)
    end

    def processar_lote(cpfs) when is_list(cpfs) do
    GenServer.call(__MODULE__, {:processar_lote, cpfs}, :infinity)
    end

    def status do
    GenServer.call(__MODULE__, :status)
    end

    # Callbacks
    @impl true
    def init(state) do
    {:ok, state}
    end

    @impl true
    def handle_call({:processar_lote, cpfs}, from, state) do
    state = %{state | pendentes: cpfs, resultados: [], em_processamento: 0}
    state = disparar_tasks(state, from)
    {:noreply, state}
    end

    @impl true
    def handle_call(:status, _from, state) do
    {:reply, %{
    pendentes: length(state.pendentes),
    em_processamento: state.em_processamento,
    concluidos: length(state.resultados)
    }, state}
    end

    @impl true
    def handle_info({ref, resultado}, state) when is_reference(ref) do
    Process.demonitor(ref, [:flush])
    state = %{state | resultados: [resultado | state.resultados], em_processamento: state.em_processamento - 1}
    state = disparar_tasks(state, nil)
    {:noreply, state}
    end

    @impl true
    def handle_info({:DOWN, _ref, :process, _pid, _reason}, state) do
    state = %{state | em_processamento: state.em_processamento - 1}
    {:noreply, state}
    end

    defp disparar_tasks(state, from) do
    slots_disponiveis = state.max_concurrent - state.em_processamento
    {a_processar, restantes} = Enum.split(state.pendentes, slots_disponiveis)

    Enum.each(a_processar, fn cpf ->
    Task.async(fn -> consultar_cpf(cpf, state.api_key) end)
    end)

    new_state = %{state |
    pendentes: restantes,
    em_processamento: state.em_processamento + length(a_processar)
    }

    if new_state.pendentes == [] and new_state.em_processamento == 0 and from do
    GenServer.reply(from, Enum.reverse(new_state.resultados))
    end

    new_state
    end

    defp consultar_cpf(cpf, api_key) do
    headers = [{"x-api-key", api_key}]
    url = "https://api.cpfhub.io/cpf/#{cpf}"

    case HTTPoison.get(url, headers, recv_timeout: 10_000) do
    {:ok, %{status_code: 200, body: body}} ->
    dados = Jason.decode!(body)
    {cpf, :ok, dados["data"]}

    {:ok, %{status_code: status}} ->
    {cpf, :erro, "Status #{status}"}

    {:error, reason} ->
    {cpf, :erro, inspect(reason)}
    end
    end
end

defmodule CpfBatchProcessor do
    @base_url "https://api.cpfhub.io"

    def processar(cpfs, api_key, opts \\ []) do
    max_concurrency = Keyword.get(opts, :max_concurrency, 5)
    timeout = Keyword.get(opts, :timeout, 15_000)

    cpfs
    |> Task.async_stream(
    fn cpf -> consultar(cpf, api_key) end,
    max_concurrency: max_concurrency,
    timeout: timeout,
    on_timeout: :kill_task
    )
    |> Enum.map(fn
    {:ok, resultado} -> resultado
    {:exit, :timeout} -> {:erro, "Timeout na consulta"}
    end)
    end

    defp consultar(cpf, api_key) do
    headers = [{"x-api-key", api_key}]

    case HTTPoison.get("#{@base_url}/cpf/#{cpf}", headers) do
    {:ok, %{status_code: 200, body: body}} ->
    %{"data" => data} = Jason.decode!(body)
    {:ok, cpf, data}

    {:ok, %{status_code: status}} ->
    {:erro, cpf, "HTTP #{status}"}

    {:error, %{reason: reason}} ->
    {:erro, cpf, reason}
    end
    end
end

defmodule CpfSupervisedProcessor do
    def processar_com_supervisor(cpfs, api_key) do
    cpfs
    |> Enum.map(fn cpf ->
    Task.Supervisor.async_nolink(CpfTaskSupervisor, fn ->
    consultar_com_retry(cpf, api_key, 3)
    end)
    end)
    |> Task.yield_many(15_000)
    |> Enum.map(fn
    {_task, {:ok, resultado}} -> resultado
    {task, nil} ->
    Task.shutdown(task, :brutal_kill)
    {:erro, "Timeout"}
    {_task, {:exit, reason}} ->
    {:erro, "Processo falhou: #{inspect(reason)}"}
    end)
    end

    defp consultar_com_retry(cpf, api_key, tentativas_restantes) do
    headers = [{"x-api-key", api_key}]
    url = "https://api.cpfhub.io/cpf/#{cpf}"

    case HTTPoison.get(url, headers) do
    {:ok, %{status_code: 200, body: body}} ->
    {:ok, cpf, Jason.decode!(body)}

    {:ok, %{status_code: status}} when status in [500, 503] and tentativas_restantes > 0 ->
    Process.sleep(1000 * (4 - tentativas_restantes))
    consultar_com_retry(cpf, api_key, tentativas_restantes - 1)

    {:error, _} when tentativas_restantes > 0 ->
    Process.sleep(500)
    consultar_com_retry(cpf, api_key, tentativas_restantes - 1)

    other ->
    {:erro, cpf, other}
    end
    end
end

# No Application supervisor
children = [
    {Task.Supervisor, name: CpfTaskSupervisor},
    {CpfProcessor, api_key: System.get_env("CPFHUB_API_KEY"), max_concurrent: 5}
]

Supervisor.start_link(children, strategy: :one_for_one)

# Uso com Task.async_stream
api_key = System.get_env("CPFHUB_API_KEY")
cpfs = ["12345678900", "98765432100", "11122233344"]

resultados = CpfBatchProcessor.processar(cpfs, api_key, max_concurrency: 3)

Enum.each(resultados, fn
    {:ok, cpf, data} ->
    IO.puts("[OK] #{cpf} -> #{data["name"]}")

    {:erro, cpf, reason} ->
    IO.puts("[ERRO] #{cpf} -> #{reason}")

    {:erro, reason} ->
    IO.puts("[ERRO] #{reason}")
end)