oban_claude

# examples/playground.exs
#
# An offline, deterministic look at oban_claude running on a *real* Oban queue.
#
# Boots an Oban instance on the SQLite (Lite) engine -- no Postgres, no Docker,
# just a throwaway file in the tmp dir -- and runs the REAL `use
# ObanClaude.Worker`. The only thing faked is the claude call itself, injected
# via the worker's `:query_fun` seam. So this exercises the genuine path:
# perform/1 -> ObanClaude.run/2 -> build/1 -> classify -> handle_result/2.
#
#   mix run examples/playground.exs

import Ecto.Query, only: [from: 2]

# ---------------------------------------------------------------------------
# 1. A throwaway SQLite-backed Ecto repo (fresh file every run).
# ---------------------------------------------------------------------------
db_path = Path.join(System.tmp_dir!(), "oban_claude_playground.db")
for suffix <- ["", "-shm", "-wal"], do: File.rm(db_path <> suffix)

defmodule PlaygroundRepo do
  use Ecto.Repo, otp_app: :oban_claude, adapter: Ecto.Adapters.SQLite3
end

Application.put_env(:oban_claude, PlaygroundRepo,
  database: db_path,
  # SQLite has a single writer, so a one-connection pool is the natural fit
  # (and avoids a connect-time "database is locked" race). busy_timeout makes
  # any contender wait rather than error.
  pool_size: 1,
  busy_timeout: 5_000,
  log: false
)

# ---------------------------------------------------------------------------
# 2. Oban's schema migration, run programmatically against the fresh DB.
#    (`Oban.Migrations.up/0` auto-detects the SQLite adapter.)
# ---------------------------------------------------------------------------
defmodule PlaygroundMigration do
  use Ecto.Migration
  def up, do: Oban.Migrations.up()
  def down, do: Oban.Migrations.down()
end

# ---------------------------------------------------------------------------
# 3. The REAL worker, with claude stubbed via :query_fun.
# ---------------------------------------------------------------------------
defmodule PlaygroundWorker do
  use ObanClaude.Worker,
    queue: :claude,
    max_attempts: 3,
    query_fun: &__MODULE__.fake_query/2,
    classifier: &__MODULE__.classify/1

  alias ClaudeWrapper.{Error, Result}

  # Tiny backoff so retries are watchable in seconds (Oban's default is ~15s+).
  @impl Oban.Worker
  def backoff(%Oban.Job{}), do: 1

  # The default classifier now maps :timeout to {:error, :timeout} (snooze
  # bypasses max_attempts, so it is no longer a default -- see ObanClaude.Outcome).
  # This worker opts back into a snooze for :timeout purely to demonstrate the
  # parked/scheduled state below; everything else defers to the default mapping.
  def classify({:error, %Error{kind: :timeout} = e}), do: {{:snooze, 30}, e}
  def classify(outcome), do: ObanClaude.Outcome.classify(outcome)

  # The worker's one override point: branch on a typed structured outcome.
  @impl ObanClaude.Worker
  def handle_result(result, _job) do
    case ObanClaude.outcome(result) do
      "blocked" -> {:cancel, :blocked}
      _ -> :ok
    end
  end

  # Stubbed claude entrypoint. The prompt doubles as the scenario selector, and
  # query_opts shows the args build/1 produced -- note `permission_mode` arrives
  # here as the atom `:bypass_permissions` even though it was enqueued as a JSON
  # string. Returns the exact shapes ClaudeWrapper.query/2 returns.
  def fake_query(prompt, query_opts) do
    IO.puts("  [claude] prompt=#{inspect(prompt)} opts=#{inspect(query_opts)}")
    stub_outcome(prompt)
  end

  defp stub_outcome("ok"),
    do: {:ok, %Result{result: "all good", is_error: false, cost_usd: 0.0123}}

  defp stub_outcome("blocked"),
    do:
      {:ok,
       %Result{
         result: "",
         is_error: false,
         extra: %{"structured_output" => %{"outcome" => "blocked"}}
       }}

  defp stub_outcome("timeout"), do: {:error, %Error{kind: :timeout}}
  defp stub_outcome("auth"), do: {:error, %Error{kind: :auth, reason: :invalid_api_key}}
  defp stub_outcome("command_failed"), do: {:error, %Error{kind: :command_failed, exit_code: 1}}
end

# ---------------------------------------------------------------------------
# 4. Boot: start the repo, migrate, start Oban on the Lite engine.
# ---------------------------------------------------------------------------
{:ok, _} = PlaygroundRepo.start_link()
Ecto.Migrator.up(PlaygroundRepo, 1, PlaygroundMigration, log: false)

{:ok, _} =
  Oban.start_link(
    repo: PlaygroundRepo,
    engine: Oban.Engines.Lite,
    notifier: Oban.Notifiers.PG,
    # Single-node script: always-leader peer so the Stager promotes
    # `retryable`/`scheduled` jobs to `available`. NOTE: `plugins: []`, not
    # `plugins: false` -- `false` is the testing switch that forces a
    # non-leader peer and turns staging off, so retries would never fire.
    peer: Oban.Peers.Isolated,
    plugins: [],
    queues: [claude: 1]
  )

# ---------------------------------------------------------------------------
# 5. Enqueue one job per claude outcome, then watch the queue resolve them.
#    The "ok" job also carries pass-through opts to show JSON -> query mapping
#    (and the permission_mode string -> atom coercion).
# ---------------------------------------------------------------------------
jobs = [
  {%{"prompt" => "ok", "model" => "sonnet", "permission_mode" => "bypass_permissions"},
   "clean result -> :ok", []},
  {%{"prompt" => "blocked"}, "structured 'blocked' -> handle_result cancels", []},
  {%{"prompt" => "timeout"}, "transient -> snooze 30s (classifier override)", []},
  {%{"prompt" => "auth"}, "config problem -> cancel (terminal)", []},
  {%{"prompt" => "command_failed"}, "infra blip -> retry then discard", [max_attempts: 2]}
]

IO.puts("\nEnqueuing #{length(jobs)} jobs on the :claude queue...\n")

id_label =
  for {args, label, opts} <- jobs, into: %{} do
    {:ok, job} = args |> PlaygroundWorker.new(opts) |> Oban.insert()
    {job.id, "#{args["prompt"]} (#{label})"}
  end

ids = Map.keys(id_label)
terminal = ~w(completed cancelled discarded scheduled)

snapshot = fn ->
  PlaygroundRepo.all(
    from(j in "oban_jobs",
      where: j.id in ^ids,
      select: {j.id, j.state, j.attempt, j.max_attempts},
      order_by: j.id
    )
  )
end

print = fn rows ->
  IO.puts("  " <> String.duplicate("-", 74))

  for {id, state, attempt, max} <- rows do
    IO.puts(
      "  ##{id}  #{String.pad_trailing(state, 10)} attempt #{attempt}/#{max}  #{id_label[id]}"
    )
  end
end

# Watch the queue tick: one compact line per poll so the transitions are
# visible (e.g. command_failed walks retryable -> available -> discarded).
# Halt once every job is terminal or parked by the snooze.
IO.puts("Watching state transitions:\n")

Enum.reduce_while(1..25, nil, fn i, _ ->
  Process.sleep(400)
  rows = snapshot.()

  line =
    rows
    |> Enum.map(fn {id, state, _, _} -> "##{id}:#{state}" end)
    |> Enum.join("  ")

  IO.puts("  t=#{Float.round(i * 0.4, 1)}s  #{line}")

  if Enum.all?(rows, fn {_, state, _, _} -> state in terminal end),
    do: {:halt, rows},
    else: {:cont, rows}
end)

IO.puts("\nFinal states:")
print.(snapshot.())

IO.puts("""

  completed  succeeded (:ok)
  cancelled  {:cancel, _}: terminal, no retry  (auth, and blocked via handle_result)
  discarded  ran out of attempts after {:error, _} retries  (command_failed)
  scheduled  {:snooze, 30}: parked ~30s out; Oban bumped max_attempts so the
             snooze did not cost a retry  (timeout, via this worker's classifier
             override -- the DEFAULT mapping returns {:error, :timeout}, since a
             snooze would let a stuck run retry forever)
""")