Petal Framework

Per-User AI Cost Tracking & Budgets

Every LLM call gets logged with input tokens, output tokens, and the price-per-million snapshot from the moment it ran. A user blasting through your free tier? You can see it in real time and shut them off.

This is the pattern Petal Pro ships with. The Req plugin attaches to your LLM client and writes an ai_call_logs row on every response. Pricing is snapshotted per-call, so an OpenAI price drop next month doesn’t retroactively rewrite last week’s history.

Pair this with a daily budget check before kicking off any expensive job.

defmodule MyApp.Repo.Migrations.CreateAiCallLogs do
  use Ecto.Migration

  def change do
    create table(:ai_call_logs, primary_key: false) do
      add :id, :uuid, primary_key: true
      add :user_id, references(:users, type: :uuid, on_delete: :nilify_all)
      add :provider, :string, null: false
      add :model, :string, null: false
      add :input_tokens, :integer, default: 0
      add :output_tokens, :integer, default: 0
      add :input_cost_per_million_cents, :integer
      add :output_cost_per_million_cents, :integer
      add :total_cost_cents, :integer, default: 0
      add :duration_ms, :integer
      add :status, :string
      add :request_body, :map
      add :response_body, :map
      timestamps(type: :utc_datetime_usec)
    end

    create index(:ai_call_logs, [:user_id, :inserted_at])
    create index(:ai_call_logs, [:provider, :model])
  end
end

defmodule MyApp.AI.Pricing do
  @moduledoc "Cents per 1M tokens. Snapshotted into ai_call_logs at call time."

  @prices %{
    "openai:gpt-4o-mini" => %{input: 15, output: 60},
    "openai:gpt-4o" => %{input: 250, output: 1000},
    "anthropic:claude-haiku-4-5-20251001" => %{input: 100, output: 500},
    "anthropic:claude-sonnet-4-6" => %{input: 300, output: 1500},
    "google:gemini-2.5-flash" => %{input: 7, output: 30},
    "perplexity:sonar-pro" => %{input: 300, output: 1500}
  }

  def for_model(model), do: Map.get(@prices, model, %{input: 0, output: 0})

  def cost_cents(model, input_tokens, output_tokens) do
    %{input: ip, output: op} = for_model(model)
    div(input_tokens * ip + output_tokens * op, 1_000_000)
  end
end

defmodule MyApp.AI.CostTrackerPlugin do
  @moduledoc "Req plugin that logs every LLM response into ai_call_logs."

  alias MyApp.AI.{AICallLog, Pricing}
  alias MyApp.Repo

  def attach(%Req.Request{} = req, opts \\ []) do
    req
    |> Req.Request.register_options([:cost_tracker_user_id, :cost_tracker_model])
    |> Req.Request.merge_options(opts)
    |> Req.Request.append_response_steps(cost_tracker: &log_response/1)
  end

  defp log_response({req, resp}) do
    model = req.options[:cost_tracker_model]
    user_id = req.options[:cost_tracker_user_id]

    with %{"usage" => %{"prompt_tokens" => in_t, "completion_tokens" => out_t}} <- resp.body do
      %{input: ip, output: op} = Pricing.for_model(model)

      %AICallLog{}
      |> AICallLog.changeset(%{
        user_id: user_id,
        provider: model |> String.split(":") |> hd(),
        model: model,
        input_tokens: in_t,
        output_tokens: out_t,
        input_cost_per_million_cents: ip,
        output_cost_per_million_cents: op,
        total_cost_cents: Pricing.cost_cents(model, in_t, out_t),
        status: "success",
        request_body: req.body,
        response_body: resp.body
      })
      |> Repo.insert!()
    end

    {req, resp}
  end
end

defmodule MyApp.AI.Budgets do
  @moduledoc "Daily per-user budget enforcement."

  import Ecto.Query
  alias MyApp.AI.AICallLog
  alias MyApp.Repo

  @daily_cap_cents 50

  def daily_cost_cents(user_id) do
    since = DateTime.utc_now() |> DateTime.add(-1, :day)

    AICallLog
    |> where([l], l.user_id == ^user_id and l.inserted_at >= ^since)
    |> Repo.aggregate(:sum, :total_cost_cents)
    |> Kernel.||(0)
  end

  def within_budget?(user_id) do
    daily_cost_cents(user_id) < @daily_cap_cents
  end

  def check!(user_id) do
    if within_budget?(user_id) do
      :ok
    else
      {:error, :daily_budget_exceeded}
    end
  end
end

defmodule MyApp.AI.Chat do
  alias MyApp.AI.{Budgets, CostTrackerPlugin}

  def respond(user, prompt) do
    with :ok <- Budgets.check!(user.id) do
      req =
        Req.new(base_url: "https://api.openai.com/v1")
        |> Req.Request.put_header("authorization", "Bearer " <> System.get_env("OPENAI_API_KEY"))
        |> CostTrackerPlugin.attach(
          cost_tracker_user_id: user.id,
          cost_tracker_model: "openai:gpt-4o-mini"
        )

      Req.post(req,
        url: "/chat/completions",
        json: %{
          model: "gpt-4o-mini",
          messages: [%{role: "user", content: prompt}]
        }
      )
    end
  end
end

Why snapshot pricing per-call?

Providers cut prices regularly. If you derive cost from a current price table, your historical revenue/margin numbers change every time OpenAI does a price drop. You don't want that.
Storing input_cost_per_million_cents alongside input_tokens means yesterday's row is correct forever.
Pair with a per-user daily cap to stop one abusive user from emptying your AI budget overnight.