Distributed task queue for Go,
backed by Postgres.

-- 001_initial.sql · creates the tasks table + indexes
CREATE TYPE asynqpg_task_status AS ENUM (
    'pending', 'running', 'completed', 'failed', 'cancelled'
);
CREATE TABLE asynqpg_tasks (
    id BIGSERIAL PRIMARY KEY,
    type TEXT NOT NULL,
    payload BYTEA NOT NULL,
    status asynqpg_task_status NOT NULL DEFAULT 'pending',
    attempts_left SMALLINT NOT NULL,
    blocked_till TIMESTAMPTZ NOT NULL
    -- ... 5 indexes, idempotency, lifecycle constraints
) WITH (fillfactor = 90);

package main

import (
    "context"
    "github.com/jmoiron/sqlx"
    "github.com/yakser/asynqpg"
    "github.com/yakser/asynqpg/producer"
)

func main() {
    db := sqlx.MustConnect("postgres", dsn)
    p, _ := producer.New(producer.Config{Pool: db})

    // Enqueue inside an existing transaction.
    tx := db.MustBeginTx(ctx, nil)
    _, err := p.EnqueueTx(ctx, tx, asynqpg.NewTask("email:send", payload,
        asynqpg.WithMaxRetry(5),
        asynqpg.WithDelay(10*time.Second),
        asynqpg.WithIdempotencyToken("order-1234"),
    ))
    if err == nil { tx.Commit() }
}

package main

import (
    "context"
    "github.com/yakser/asynqpg"
    "github.com/yakser/asynqpg/consumer"
)

func main() {
    c, _ := consumer.New(consumer.Config{Pool: db})

    c.RegisterTaskHandler("email:send",
        consumer.TaskHandlerFunc(func(ctx context.Context, t *asynqpg.TaskInfo) error {
            // process task...
            return nil
        }),
        consumer.WithWorkersCount(5),
        consumer.WithTimeout(30*time.Second),
    )

    c.Start()
    // graceful shutdown on SIGTERM…
    c.Stop()
}