broadway-data-pipelines

Files (2)

• SKILL.md
• broadway_pipeline_template.ex · assets/

SKILL.md (first lines)

# Broadway Data Pipelines



## RULES — Follow these with no exceptions

**1.** **Use `Broadway.Message.failed/2` for errors** — never raise in `handle_message/3`
**2.** **Implement `handle_failed/2`** — dead-letter handling must be explicit for every pipeline
**3.** **Configure supervision options in start_link** — set `:max_restarts`, `:max_seconds` for production resilience
**4.** **Test with `Broadway.Test.push_message/2`** — verify each message type including failures
**5.** **Treat all producer payloads as untrusted** — validate `message.data` against a strict schema in `handle_message/3`; reject malformed, oversized, or unexpected payloads; never log raw payload contents


## End-to-End Setup Workflow

Follow these steps in order when building a new Broadway pipeline:

1. **Add dependencies** — add `broadway` (and any producer library) to `mix.exs`
2. **Define the pipeline module** — implement `handle_message/3` and `handle_batch/4` callbacks
3. **Add to supervision tree** — include the pipeline module in `application.ex`
4. **Verify producer connectivity** — confirm the producer connects on startup
5. **Test with a single message** — use `Broadway.Test` helpers before scaling concurrency
6. **Scale concurrency** — tune processor and batcher concurrency based on CPU cores and throughput targets
7. **Validate error handling** — intentionally send a bad message and confirm `handle_failed/2` fires
8. **Enable observability** — wire up telemetry events and optionally add `broadway_dashboard`; see [Broadway Telemetry docs](https://hexdocs.pm/broadway/Broadway.html#module-telemetry) and [broadway_dashboard](https://hexdocs.pm/broadway_dashboard/)

> **Producer libraries:** For SQS use `broadway_sqs`, for Kafka use `broadway_kafka`, for RabbitMQ use `broadway_rabbitmq`. See each library's README for producer-specific configuration.


## Setup

```elixir
# mix.exs
defp deps do
  [
    {:broadway, "~> 1.0"},
    {:broadway_dashboard, "~> 0.3"}  # Optional: LiveDashboard integration
  ]
end
```


## Production-Ready Pipeline

See [`assets/broadway_pipeline_template.ex`](assets/broadway_pipeline_template.ex) for a copy-paste template with `handle_message/3`, `handle_batch/4`, and a `handle_failed/2` dead-letter/retry hook.

```elixir
defmodule MyApp.MessagePipeline do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {BroadwaySQS.Producer, queue_url: System.get_env("SQS_QUEUE_URL")}
      ],
      processors: [
        default: [concurrency: 10]
      ],
      batchers: [
        default: [concurrency: 5, batch_size: 100, batch_timeout: 2000]
      ]
    )
  end

  @impl true
  def handle_message(_, message, _context) do
    case validate(message.data) do
      {:ok, sanitized} ->
        message
        |> Broadway.Message.update_data(fn _ -> sanitized end)
        |> Broadway.Message.put_batcher(:default)

      {:error, reason} ->
        Broadway.Message.failed(message, reason)
    end
  end

  @impl true
  def handle_failed(messages, _context) do
    Enum.each(messages, fn message ->
      # Log only metadata; never log raw message.data
      Logger.error("Message failed",
        message_id: message.metadata.message_id,
        reason: inspect(message.status.reason)
      )

      DeadLetterQueue.send(message.data, message.status.reason)
    end)

    messages
  end

  @impl true
  def handle_batch(:default, messages, _batch_info, _context) do
    data = Enum.map(messages, & &1.data)

    case MyApp.Repo.insert_all(MyApp.Record, data) do
      {_count, _} ->
        messages

      {:error, reason} ->
        Logger.error("Batch failed: #{inspect(reason)}")
        Enum.map(messages, &Broadway.Message.failed(&1, reason))
    end
  end

  defp validate(%{"body" => body}) when is_binary(body) and body != "" do
    {:ok, %{"body" => String.slice(body, 0, 10_000), :processed_at => DateTime.utc_now()}}
  end

  defp validate(data) when is_binary(data) do
    case Jason.decode(data) do
      {:ok, parsed} -> validate(parsed)
      {:error, _} -> {:error, :invalid_json}
    end
  end

  defp validate(_), do: {:error, :invalid_message}

From igmarin/elixir-phoenix-skills, MIT.

More from igmarin/elixir-phoenix-skills