# BroadwayPgQue

Official [Broadway](https://github.com/dashbitco/broadway) producer for [PgQue](https://pgque.dev/) — a high-throughput, zero-bloat background queue engine for PostgreSQL.

---

## Features

- **Zero Table Bloat:** Integrates with PgQue's rotating partitioned tables and instant `TRUNCATE` cycle.
- **Demand-Driven Concurrency:** Native backpressure and memory safety backed by Broadway and GenStage.
- **Adaptive Idle Backoff:** Automatically backs off polling intervals when queues are empty to preserve database CPU.
- **Instant Wake-Up:** Bypasses idle polling backoffs immediately when new messages are enqueued via `BroadwayPgQue.wake_up/1`.
- **Custom Transformers:** Easily decode JSON or translate raw PgQue records into domain structs.
- **Flexible Acknowledgements:** Built-in support for PgQue batch acknowledgements, delayed retries, and dead-letter queue (DLQ) routing.
- **Cooperative Subconsumers:** Built-in support for PgQue consumer groups (`pgque.receive_coop`).

---

## Installation

Add `broadway_pgque` to your dependencies in `mix.exs`:

```elixir
def deps do
  [
    {:broadway_pgque, "~> 0.1.0"},
    {:postgrex, ">= 0.0.0"}
  ]
end
```

---

## Usage

Define a Broadway pipeline module:

```elixir
defmodule MyApp.NotificationPipeline do
  use Broadway

  alias BroadwayPgQue.Producer, as: PgQueProducer

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {
          PgQueProducer,
          repo: MyApp.Repo,
          queue: "notifications",
          poll_interval: 1000,
          max_poll_interval: 30_000,
          concurrency: 1
        }
      ],
      processors: [
        default: [concurrency: 20]
      ],
      batchers: [
        default: [batch_size: 100, batch_timeout: 2000]
      ]
    )
  end

  @impl Broadway
  def handle_message(_processor, message, _context) do
    # message.data contains the raw payload
    # message.metadata contains :id, :batch_id, :type, :retry_count, :created_at
    MyApp.Notifications.send(message.data)
    message
  end

  @impl Broadway
  def handle_batch(:default, messages, _batch_info, _context) do
    # Batch processing with automatic chunk buffering
    payloads = Enum.map(messages, & &1.data)
    MyApp.Notifications.send_bulk(payloads)
    messages
  end
end
```

### Instant Wake-Up on Enqueue

To eliminate any latency when messages are inserted into an idle queue:

```elixir
BroadwayPgQue.wake_up(MyApp.NotificationPipeline)
```

### Custom Message Transformers

Transform raw messages before Broadway processors handle them:

```elixir
defmodule MyApp.Transformer do
  def transform(raw_msg, ack_ref, ack_data, _opts) do
    decoded = Jason.decode!(raw_msg.payload)

    %Broadway.Message{
      data: decoded,
      metadata: %{id: raw_msg.msg_id, type: raw_msg.type},
      acknowledger: {ack_data.acknowledger, ack_ref, Map.put(ack_data, :msg_id, raw_msg.msg_id)}
    }
  end
end
```

And configure:

```elixir
producer: [
  module: {
    PgQueProducer,
    repo: MyApp.Repo,
    queue: "notifications",
    transformer: {MyApp.Transformer, :transform, []}
  }
]
```

---

## Configuration Reference

`BroadwayPgQue.Producer` uses [NimbleOptions](https://hexdocs.pm/nimble_options) to validate all options.

### Database & Connection Pooling Options

`BroadwayPgQue` can run **100% without Ecto**. You can choose your connection pooling module or let it auto-detect:

| Option | Type | Default | Description |
| :--- | :--- | :--- | :--- |
| `:pool` | `module` | `nil` (auto) | Module implementing `BroadwayPgQue.Pool` (`BroadwayPgQue.Pool.Postgrex`, `BroadwayPgQue.Pool.Ecto`, or custom). |
| `:conn` | `pid` \| `atom` | *None* | Direct Postgrex connection process PID or registered name (**no Ecto required**). |
| `:connection_opts` | `keyword` | *None* | Options passed to `Postgrex.start_link/1` to spin up a dedicated connection/pool (**no Ecto required**). |
| `:repo` | `atom` | *None* | Shared `Ecto.Repo` module. |
| `:pool_opts` | `keyword` | `[]` | Options passed to custom `pool.init/1` callback. |
| `:query_timeout` | `timeout` | `15_000` | Query timeout in milliseconds for PgQue SQL operations. |

#### 1. Pure Postgrex (No Ecto Required!)

To run BroadwayPgQue without having Ecto in your project at all:

```elixir
# Using an existing Postgrex process or pool
producer: [
  module: {
    BroadwayPgQue.Producer,
    pool: BroadwayPgQue.Pool.Postgrex,
    conn: :my_postgrex_pool, # or conn pid
    queue: "orders"
  }
]

# Or letting BroadwayPgQue manage its own dedicated Postgrex connection:
producer: [
  module: {
    BroadwayPgQue.Producer,
    queue: "orders",
    connection_opts: [
      hostname: "localhost",
      database: "my_db",
      username: "postgres",
      pool_size: 2
    ]
  }
]
```

#### 2. Writing a Custom Pool Module

Implement the `BroadwayPgQue.Pool` behaviour to plug in your own connection pool, telemetry wrapper, or test mock:

```elixir
defmodule MyApp.CustomPool do
  @behaviour BroadwayPgQue.Pool

  @impl true
  def init(opts) do
    # initialize or resolve connection state
    {:ok, opts[:conn_name]}
  end

  @impl true
  def query(conn_name, sql, params, opts) do
    Postgrex.query(conn_name, sql, params, opts)
  end
end
```

### Polling & Scheduling Options

| Option | Type | Default | Description |
| :--- | :--- | :--- | :--- |
| `:poll_type` | `:fixed` \| `:adaptive` | `:fixed` | Polling strategy when queue is empty: `:fixed` (constant `poll_interval`) or `:adaptive` (exponential backoff up to `max_poll_interval`). |
| `:poll_interval` | `pos_integer` | `1000` | Base polling interval in milliseconds when queue is empty. |
| `:max_poll_interval` | `pos_integer` | `30_000` | Maximum exponential backoff interval in milliseconds. |
| `:backoff_factor` | `float` | `1.5` | Multiplier applied when consecutive polls yield no messages. |
| `:max_batch_size` | `pos_integer` | `100` | Maximum messages pulled per batch query. |
| `:receive_mode` | `:auto` \| `:standard` \| `:coop` | `:auto` | Uses `:coop` (`pgque.receive_coop`) if `:subconsumer` is provided, `:standard` otherwise. |
| `:ticker_max_lag` | `string` | `"0.1 seconds"` | PostgreSQL setting for queue ticker lag. |

### Acknowledgement & Error Handling Options

| Option | Type | Default | Description |
| :--- | :--- | :--- | :--- |
| `:acknowledger` | `atom` | `BroadwayPgQue.Acknowledger` | Broadway acknowledger module. |
| `:on_failure` | `:dead` \| `:retry` \| `function` | `:dead` | Action for unhandled failed messages: move to DLQ (`:dead`) or retry (`:retry`). |
| `:default_retry_delay` | `pos_integer` | `60` | Delay in seconds when scheduling failed messages for retry. |
| `:transformer` | `mfa` \| `function` | `{BroadwayPgQue.Producer, :transform, []}` | Callback to transform raw database maps into `Broadway.Message`. |

### Telemetry Events

`BroadwayPgQue` emits standard `:telemetry` events for visibility:

- `[:broadway_pgque, :fetch, :start]` — `%{system_time: integer()}`, `%{queue: String.t(), consumer: String.t(), demand: integer()}`
- `[:broadway_pgque, :fetch, :stop]` — `%{duration: integer(), count: integer()}`, `%{queue: String.t(), consumer: String.t(), batch_id: integer()}`
- `[:broadway_pgque, :ack, :start]` — `%{system_time: integer()}`, `%{batch_id: integer(), queue: String.t(), total_count: integer(), failed_count: integer()}`
- `[:broadway_pgque, :ack, :stop]` — `%{duration: integer()}`, `%{batch_id: integer(), queue: String.t()}`
- `[:broadway_pgque, :ack, :exception]` — `%{duration: integer()}`, `%{batch_id: integer(), queue: String.t(), error: term()}`

---

## License

Apache License 2.0.