fleet

# Fleet

High-throughput, zero-bloat background job processing engine for Elixir, powered by **PgQue** and **Broadway**.

---

## Key Highlights

- **Zero-Bloat Architecture:** Powered by [PgQue](https://pgque.dev/) using snapshot-based batching and rotating tables. Reclaiming disk space happens via instant `TRUNCATE`, completely eliminating table and index dead-tuple bloat under heavy workloads.
- **Physical Queue Isolation:** Each queue is provisioned with its own physical partition tables in PostgreSQL. A traffic surge on an analytics queue will never contend for row locks or degrade a critical mailer queue.
- **Demand-Driven Concurrency:** Backed by Broadway and GenStage, providing native backpressure, graceful shutdown, and memory protection.
- **Transactional Enqueueing (Outbox):** Enqueue jobs within the same Ecto transaction as your business records (`Ecto.Multi` support included).
- **Native Batching (Chunks):** Built-in support for chunked batch execution (`perform_batch/1`) for bulk external operations like API requests or data ingestion.
- **Configurable History & Retention:** Configurable ring-buffer partitions (`history_tables: 7, rotation_period: "1 day"`) allow querying completed jobs for days without extra audit tables.
- **Internal BEAM Ticker:** Built-in scheduler process allows running test suites and local development without mandatory `pg_cron` setup.
- **Declarative Cron:** Built-in recurring job scheduler supporting cron expressions and interval statements (`@every 30s`, `@hourly`, etc.).

---

## Installation

Add `fleet` to your list of dependencies in `mix.exs`:

```elixir
def deps do
  [
    {:fleet, "~> 0.1.0"}
  ]
end
```

### 1. Run the Migration

Generate a migration in your application:

```bash
mix ecto.gen.migration add_fleet
```

And invoke `Fleet.Migration`:

```elixir
defmodule MyApp.Repo.Migrations.AddFleet do
  use Ecto.Migration

  def up do
    Fleet.Migration.up()
  end

  def down do
    Fleet.Migration.down()
  end
end
```

Run `mix ecto.migrate` to create the schema and functions.

---

## Defining Workers

Define background workers by using `Fleet.Worker`:

```elixir
defmodule MyApp.Workers.SendEmail do
  use Fleet.Worker, queue: :mailers, max_attempts: 5

  @impl Fleet.Worker
  def perform(%Fleet.Job{args: %{"to" => to, "subject" => subject}}) do
    MyApp.Mailer.send(to, subject)
  end
end
```

### Strongly Typed Workers with `Fleet.Schema`

For compile-time type safety, automatic runtime casting, and changeset validation, use `Fleet.Schema`:

```elixir
defmodule MyApp.Workers.SendInvoice do
  use Fleet.Worker, queue: :billing, max_attempts: 3
  use Fleet.Schema

  embedded_schema do
    field :invoice_id, :integer
    field :amount, :decimal
    field :currency, :string, default: "USD"
  end

  # Optional: custom validation rules
  def changeset(schema, attrs) do
    schema
    |> cast(attrs, [:invoice_id, :amount, :currency])
    |> validate_required([:invoice_id, :amount])
  end

  @impl Fleet.Worker
  def perform(%Fleet.Job{args: %__MODULE__{invoice_id: id, amount: amount, currency: curr}}) do
    # 100% typed: `amount` is Decimal, `id` is integer, `curr` defaults to "USD"
    MyApp.Billing.charge(id, amount, curr)
  end
end
```

#### Enqueueing Typed Jobs

```elixir
# Auto-casts strings to Decimals, integers, DateTimes, etc. Raises on invalid changeset
job = MyApp.Workers.SendInvoice.new(%{invoice_id: 101, amount: "49.99"})
Fleet.insert(MyApp.Repo, job)

# Safe constructor returning {:ok, job} | {:error, changeset} (ideal for web controllers)
case MyApp.Workers.SendInvoice.build(params) do
  {:ok, job} -> Fleet.insert(MyApp.Repo, job)
  {:error, %Ecto.Changeset{} = cs} -> {:error, cs}
end
```

### Snoozing and Retries

- **Retries:** If a worker raises an error or returns `{:error, reason}`, Fleet automatically schedules a retry using exponential backoff with jitter (`backoff/1`). Once attempts exceed `max_attempts`, the job is automatically moved to the Dead-Letter Queue (DLQ).
- **Snoozing:** To delay execution dynamically, return `{:snooze, seconds}`:

```elixir
@impl Fleet.Worker
def perform(%Fleet.Job{args: %{"task_id" => id}}) do
  if MyApp.TaskService.ready?(id) do
    MyApp.TaskService.run(id)
  else
    # Delay for 60 seconds
    {:snooze, 60}
  end
end
```

### Batch Workers (Chunks)

Process multiple jobs together in a single batch, automatically buffered by Broadway according to `batch_size` and `batch_timeout`:

```elixir
defmodule MyApp.Workers.BulkEvents do
  use Fleet.Worker, queue: :events

  @impl Fleet.Worker
  def perform(_job), do: :ok

  @impl Fleet.Worker
  def perform_batch(jobs) do
    payloads = Enum.map(jobs, fn job -> job.args end)
    MyApp.Analytics.send_bulk(payloads)
    :ok
  end
end
```

#### High-Performance Database Ingestion (with `ecto_unnest`)

When your batch worker needs to bulk-insert processed jobs into your application's Ecto schemas, pairing `perform_batch/1` with [`ecto_unnest`](https://hex.pm/packages/ecto_unnest) produces a zero-bloat, constant-SQL ingestion pipeline:

```elixir
defmodule MyApp.Workers.ImportUsers do
  use Fleet.Worker, queue: :imports

  @impl Fleet.Worker
  def perform_batch(jobs) do
    # Extract records buffered by Broadway
    user_ids = Enum.map(jobs, & &1.args["user_id"])
    emails   = Enum.map(jobs, & &1.args["email"])
    names    = Enum.map(jobs, & &1.args["name"])

    # Constant SQL bulk insert using PostgreSQL unnest()
    EctoUnnest.insert_all(
      MyApp.Repo,
      MyApp.User,
      %{user_id: user_ids, email: emails, name: names},
      placeholders: %{inserted_at: DateTime.utc_now()}
    )

    :ok
  end
end
```

Broadway provides the memory protection and backpressure, while `unnest()` avoids prepared statement proliferation and eliminates SQL parameter limit bottlenecks.

---

## Enqueueing Jobs

### Direct Insertion

```elixir
%{to: "jane@example.com", subject: "Welcome"}
|> MyApp.Workers.SendEmail.new()
|> Fleet.insert(MyApp.Repo)
```

### Transactional Insertion with `Ecto.Multi`

Ensure jobs are only enqueued if your business transaction commits:

```elixir
Ecto.Multi.new()
|> Ecto.Multi.insert(:user, user_changeset)
|> Fleet.insert(:welcome_email, fn %{user: user} ->
  MyApp.Workers.SendEmail.new(%{"to" => user.email, "subject" => "Welcome!"})
end)
|> MyApp.Repo.transaction()
```

### Bulk Enqueueing (Native PostgreSQL `unnest` Batching)

Enqueue hundreds or thousands of jobs in a single round-trip:

```elixir
jobs = Enum.map(users, fn user ->
  MyApp.Workers.SendEmail.new(%{"to" => user.email})
end)

Fleet.insert_all(MyApp.Repo, jobs)
```

> [!TIP]
> **Zero Statement Bloat:** Under the hood, `Fleet.insert_all/2` delegates to `pgque.send_batch/3`, which uses PostgreSQL's native `unnest($2::text[]) WITH ORDINALITY`. The query text remains 100% constant regardless of batch size, making it fully compatible with transaction-mode connection poolers like PgBouncer and completely immune to PostgreSQL's 65,535 query parameter cap.

---

## Configuration and Supervision

Add `Fleet` to your application supervision tree in `lib/my_app/application.ex`:

```elixir
def start(_type, _args) do
  children = [
    MyApp.Repo,
    {Fleet,
     repo: MyApp.Repo,
     run_ticker: true,
     queues: [
       default: [concurrency: 10],
       mailers: [concurrency: 20, poll_interval: 50],
       events: [concurrency: 5, batch_size: 50, batch_timeout: 1000]
     ],
     cron: [
       {"@hourly", MyApp.Workers.CleanupWorker},
       {"@every 30s", MyApp.Workers.HeartbeatWorker, args: %{"scope" => "all"}}
     ]}
  ]

  opts = [strategy: :one_for_one, name: MyApp.Supervisor]
  Supervisor.start_link(children, opts)
end
```

---

## Recipes & Architectural Patterns

### Recipe 1: Zero-Bloat Recurring & Mid-Month Billing (The Reconciliation Sweep)

A common challenge in subscription systems is billing customers periodically (e.g., monthly) while supporting signups that occur on arbitrary days throughout the month.

> [!WARNING]
> **Anti-Pattern: 30-Day Sleeping Queue Jobs**
> Enqueueing a background job scheduled 30 days in the future for every subscriber is an architectural anti-pattern:
> 1. **Dead-Tuple Bloat:** Delayed jobs sitting in database tables cannot leverage rotating table `TRUNCATE` operations; they accumulate dead tuples on `DELETE` and cause index bloat on scheduled timestamps.
> 2. **Stale / Zombie State:** If a subscriber cancels, upgrades, or updates their credit card on day 15, an enqueued job becomes a "zombie" that must either be manually searched and deleted or defended with complex pre-execution checks.

#### The Recommended Pattern: State in Postgres + Scheduled Sweep

Instead of storing scheduling state in queue tables, store the next renewal timestamp in your application database (`subscriptions.next_billing_at`), index it with a partial index, and run an `@hourly` or `@daily` reconciliation sweep using `Fleet.Cron`.

##### 1. Database Schema with Partial Index

In your PostgreSQL migration:

```elixir
create table(:subscriptions) do
  add :user_id, references(:users), null: false
  add :status, :string, default: "active", null: false
  add :next_billing_at, :utc_datetime, null: false
  add :amount, :decimal, precision: 10, scale: 2, null: false
  timestamps()
end

# Ultra-compact, fast partial index:
create index(:subscriptions, [:next_billing_at], where: "status = 'active'")
```

##### 2. Handling Mid-Month Signups

When a user signs up on any day of the month (e.g. October 14th), charge their initial invoice immediately and schedule their `next_billing_at` 30 days into the future within the same transaction:

```elixir
def register_and_subscribe(user_params, card_token) do
  Ecto.Multi.new()
  |> Ecto.Multi.insert(:user, User.changeset(user_params))
  |> Ecto.Multi.insert(:subscription, fn %{user: user} ->
    Subscription.changeset(%{
      user_id: user.id,
      status: "active",
      next_billing_at: DateTime.add(DateTime.utc_now(), 30, :day)
    })
  end)
  # Enqueue immediate first charge (or process synchronously):
  |> Fleet.insert(:initial_charge, fn %{subscription: sub} ->
    MyApp.Workers.ChargeSubscription.new(%{"subscription_id" => sub.id, "initial" => true})
  end)
  |> MyApp.Repo.transaction()
end
```

##### 3. The Hourly Reconciliation Sweep Worker

Use `Fleet.Cron` in your application supervision tree:

```elixir
# lib/my_app/application.ex
{Fleet,
 repo: MyApp.Repo,
 cron: [
   {"@hourly", MyApp.Workers.BillingSweepWorker}
 ]}
```

The sweep queries only active subscriptions due right now and enqueues **instant** jobs in bulk:

```elixir
defmodule MyApp.Workers.BillingSweepWorker do
  use Fleet.Worker, queue: :billing

  import Ecto.Query

  @impl Fleet.Worker
  def perform(_job) do
    now = DateTime.utc_now()

    # Query only active subscriptions that are due
    due_subs =
      MyApp.Repo.all(
        from s in MyApp.Subscription,
          where: s.status == "active" and s.next_billing_at <= ^now,
          limit: 1000
      )

    # Bulk-enqueue immediate jobs with constant SQL unnest
    jobs =
      Enum.map(due_subs, fn sub ->
        MyApp.Workers.ChargeSubscription.new(%{"subscription_id" => sub.id})
      end)

    {:ok, _} = Fleet.insert_all(MyApp.Repo, jobs)
    :ok
  end
end
```

##### 4. The Charge Worker

The worker executes under Broadway's concurrency control, charges the payment gateway, and advances `next_billing_at`:

```elixir
defmodule MyApp.Workers.ChargeSubscription do
  use Fleet.Worker, queue: :billing, max_attempts: 3

  @impl Fleet.Worker
  def perform(%Fleet.Job{args: %{"subscription_id" => sub_id}}) do
    case MyApp.Billing.charge_and_renew(sub_id) do
      :ok ->
        :ok

      {:error, :card_declined} ->
        # Mark subscription past_due; do not retry
        MyApp.Billing.mark_past_due(sub_id)
        :ok

      {:error, :gateway_timeout} ->
        # Transient failure: Fleet automatically retries with exponential backoff
        {:error, :gateway_timeout}
    end
  end
end
```

**Why this preserves Fleet's Zero-Bloat guarantees:**
* **Instant Reclaim:** Jobs exist in queue tables only during active execution (milliseconds to seconds); finished jobs cycle through rotating tables and are cleaned up with zero-bloat `TRUNCATE`.
* **Zero Zombie Jobs:** Cancellations are as simple as setting `status: "cancelled"` in the `subscriptions` table. No dangling jobs to find and cancel in queue tables.
* **Extreme Scale:** Even with millions of users, PostgreSQL's partial index (`WHERE status = 'active'`) returns due records in microseconds with minimal RAM overhead.

---

## Ring-Buffer Retention & Historical Queries

Fleet retains completed jobs in circular partition tables according to each queue's configured retention. You can query past jobs directly with `Fleet.History`:

```elixir
# Count jobs created in the last 24 hours
since = DateTime.add(DateTime.utc_now(), -86400, :second)
{:ok, count} = Fleet.History.count_jobs(MyApp.Repo, :mailers, since: since)

# Paginate recent historical events
{:ok, recent_jobs} = Fleet.History.list_jobs(MyApp.Repo, :mailers, limit: 25, offset: 0)

# Fetch a single historical job by ID
{:ok, job} = Fleet.History.get_job(MyApp.Repo, :mailers, 1024)
```

---

## Telemetry Events

Fleet emits standard `:telemetry` events:

- `[:fleet, :job, :start]` — `%{system_time: integer()}`, `%{job: Fleet.Job.t()}`
- `[:fleet, :job, :stop]` — `%{duration: integer()}`, `%{job: Fleet.Job.t(), status: :ok | :error | :snoozed}`
- `[:fleet, :job, :exception]` — `%{duration: integer()}`, `%{job: Fleet.Job.t(), kind: atom(), reason: term(), stacktrace: list()}`
- `[:fleet, :cron, :trigger]` — `%{system_time: integer()}`, `%{worker: module(), job: Fleet.Job.t()}`

---

## License

Apache License 2.0.