Getting Started
View SourceInstallation
Erlang (rebar3)
Add shigoto to your rebar.config dependencies:
{deps, [
{shigoto, {git, "https://github.com/Taure/shigoto.git", {branch, "main"}}}
]}.Elixir (Mix)
Add shigoto to your mix.exs:
defp deps do
[
{:shigoto, github: "Taure/shigoto", branch: "main"}
]
endConfiguration
Erlang (sys.config)
{shigoto, [
{pool, my_app_db},
{queues, [{<<"default">>, 10}, {<<"mailers">>, 5}]},
{poll_interval, 5000},
{prune_after_days, 14},
{shutdown_timeout, 15000},
{cron, []}
]}Elixir (config.exs)
config :shigoto,
pool: :my_app_db,
queues: [{"default", 10}, {"mailers", 5}],
poll_interval: 5000,
prune_after_days: 14,
shutdown_timeout: 15000,
cron: []Configuration Reference
| Key | Default | Description |
|---|---|---|
pool | required | Your pgo pool name |
queues | [{<<"default">>, 10}] | Queue names with concurrency limits |
poll_interval | 5000 | Milliseconds between polling for new jobs |
prune_after_days | 14 | Days before completed/discarded jobs are archived |
shutdown_timeout | 15000 | Milliseconds to wait for in-flight jobs during shutdown |
cron | [] | Cron entries (see the Cron guide) |
middleware | [] | Global middleware chain (see Middleware) |
encryption_key | undefined | 32-byte AES-256-GCM key for encrypting job args |
heartbeat_interval | 30000 | Heartbeat interval in ms for executing jobs |
load_shedding | undefined | Seki load shedding config (see Resilience) |
queue_weights | #{} | Queue weight map for weighted polling |
fair_queues | [] | Queue names that use partition-key fair claiming |
fanout_queues | [] | Broadcast queues where all nodes process every job (see Fanout Queues) |
Run the Migration
Create the necessary tables:
shigoto_migration:up(my_app_db).:shigoto_migration.up(:my_app_db)Create a Worker
Erlang
-module(my_email_worker).
-behaviour(shigoto_worker).
-export([perform/1]).
perform(#{<<"to">> := To, <<"subject">> := Subject, <<"body">> := Body}) ->
my_mailer:send(To, Subject, Body),
ok.Elixir
defmodule MyEmailWorker do
@behaviour :shigoto_worker
@impl true
def perform(%{"to" => to, "subject" => subject, "body" => body}) do
MyMailer.send(to, subject, body)
:ok
end
endNote: Job args always have binary keys (
<<"key">>/"key") because they are stored as JSONB in PostgreSQL.
Insert a Job
Erlang
shigoto:insert(#{
worker => my_email_worker,
args => #{<<"to">> => <<"user@example.com">>, <<"subject">> => <<"Welcome">>}
}).Elixir
:shigoto.insert(%{
worker: MyEmailWorker,
args: %{"to" => "user@example.com", "subject" => "Welcome"}
})With options:
:shigoto.insert(%{
worker: MyEmailWorker,
args: %{"to" => "user@example.com"},
queue: "mailers",
priority: 5,
max_attempts: 10,
tags: ["email", "onboarding"],
scheduled_at: {{2026, 3, 19}, {10, 0, 0}}
})Bulk Insert
Insert many jobs in a single SQL roundtrip:
shigoto:insert_all([
#{worker => my_worker, args => #{<<"id">> => 1}},
#{worker => my_worker, args => #{<<"id">> => 2}},
#{worker => my_worker, args => #{<<"id">> => 3}}
]).:shigoto.insert_all([
%{worker: MyWorker, args: %{"id" => 1}},
%{worker: MyWorker, args: %{"id" => 2}},
%{worker: MyWorker, args: %{"id" => 3}}
])Transactional Enqueue
If shigoto uses the same pgo pool as your application, job inserts participate in your database transactions. The job is only enqueued if the transaction commits:
pgo:transaction(fun() ->
pgo:query(~"INSERT INTO users (name, email) VALUES ($1, $2)", [Name, Email]),
shigoto:insert(#{worker => welcome_email_worker, args => #{<<"email">> => Email}})
end).If the transaction rolls back (e.g. a unique constraint violation on the user), the job is never inserted. No "user created but email lost" or "email sent but user creation failed" bugs.
This works with Kura too — just configure shigoto to use the same pool:
{pgo, [{pools, [{default, #{...}}]}]},
{shigoto, [{pool, default}]}Testing with drain_queue
Process all pending jobs synchronously in tests:
shigoto:drain_queue(<<"default">>).
shigoto:drain_queue(<<"default">>, #{timeout => 30000}).:shigoto.drain_queue("default")
:shigoto.drain_queue("default", %{timeout: 30000})Next Steps
- Workers — Error handling, retries, backoff, timeouts, and optional callbacks
- Cron — Scheduled recurring jobs
- Batches — Group jobs with completion callbacks
- Middleware — Before/after hooks for job execution
- Fanout Queues — Broadcast delivery to all nodes (cloud-native pub/sub)
- Resilience — Rate limiting, circuit breaking, and load shedding via seki