Packages
gabsurd

0.1.0
A Gleam SDK for the Absurd durable workflow system, with type-safe database access via Parrot and OTP worker actors
HexDocs
Current section
Files
Jump to
Readme 1 Version 8 Dependencies Dependants Files Activity
gabsurd src gabsurd@task.erl
Raw
src/gabsurd@task.erl

-module(gabsurd@task).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/gabsurd/task.gleam").
-export([new_options/0, with_max_attempts/2, with_retry_strategy/2, with_cancellation/2, with_headers/2, with_idempotency_key/2, encode_options/1, spawn/5, claim/5, complete/4, fail/4, fail_with_retry/5, extend_claim/4, schedule_run/4, cancel/3, get_result/3, retry/4]).
-export_type([spawn_info/0, claim/0, task_result/0, spawn_options/0, retry_strategy/0, cancellation/0]).

-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.

?MODULEDOC(
    " Task lifecycle operations for the Absurd durable workflow system.\n"
    " Provides high-level functions for spawning, claiming, completing,\n"
    " failing, and cancelling tasks.\n"
).

-type spawn_info() :: {spawn_info,
        bitstring(),
        bitstring(),
        integer(),
        boolean()}.

-type claim() :: {claim,
        bitstring(),
        bitstring(),
        integer(),
        binary(),
        binary(),
        binary(),
        integer(),
        binary(),
        binary(),
        binary()}.

-type task_result() :: {task_result, binary(), binary(), binary()}.

-type spawn_options() :: {spawn_options,
        gleam@option:option(integer()),
        gleam@option:option(retry_strategy()),
        gleam@option:option(cancellation()),
        gleam@option:option(gleam@json:json()),
        gleam@option:option(binary())}.

-type retry_strategy() :: {fixed_retry, integer()} |
    {exponential_retry, integer(), float(), gleam@option:option(float())}.

-type cancellation() :: {cancellation, integer()}.

-file("src/gabsurd/task.gleam", 92).
?DOC(" Create empty spawn options (all fields default to absent).\n").
-spec new_options() -> spawn_options().
new_options() ->
    {spawn_options, none, none, none, none, none}.

-file("src/gabsurd/task.gleam", 103).
?DOC(" Set the maximum number of attempts for this task.\n").
-spec with_max_attempts(spawn_options(), integer()) -> spawn_options().
with_max_attempts(Options, Max) ->
    {spawn_options,
        {some, Max},
        erlang:element(3, Options),
        erlang:element(4, Options),
        erlang:element(5, Options),
        erlang:element(6, Options)}.

-file("src/gabsurd/task.gleam", 108).
?DOC(" Set the retry strategy for failed tasks.\n").
-spec with_retry_strategy(spawn_options(), retry_strategy()) -> spawn_options().
with_retry_strategy(Options, Strategy) ->
    {spawn_options,
        erlang:element(2, Options),
        {some, Strategy},
        erlang:element(4, Options),
        erlang:element(5, Options),
        erlang:element(6, Options)}.

-file("src/gabsurd/task.gleam", 116).
?DOC(" Set the cancellation policy.\n").
-spec with_cancellation(spawn_options(), cancellation()) -> spawn_options().
with_cancellation(Options, Cancellation) ->
    {spawn_options,
        erlang:element(2, Options),
        erlang:element(3, Options),
        {some, Cancellation},
        erlang:element(5, Options),
        erlang:element(6, Options)}.

-file("src/gabsurd/task.gleam", 124).
?DOC(" Set headers (arbitrary JSON metadata).\n").
-spec with_headers(spawn_options(), gleam@json:json()) -> spawn_options().
with_headers(Options, Headers) ->
    {spawn_options,
        erlang:element(2, Options),
        erlang:element(3, Options),
        erlang:element(4, Options),
        {some, Headers},
        erlang:element(6, Options)}.

-file("src/gabsurd/task.gleam", 132).
?DOC(" Set an idempotency key to prevent duplicate task creation.\n").
-spec with_idempotency_key(spawn_options(), binary()) -> spawn_options().
with_idempotency_key(Options, Key) ->
    {spawn_options,
        erlang:element(2, Options),
        erlang:element(3, Options),
        erlang:element(4, Options),
        erlang:element(5, Options),
        {some, Key}}.

-file("src/gabsurd/task.gleam", 191).
-spec encode_cancellation(cancellation()) -> gleam@json:json().
encode_cancellation(C) ->
    case C of
        {cancellation, Max_duration} ->
            gleam@json:object(
                [{<<"max_duration"/utf8>>, gleam@json:int(Max_duration)}]
            )
    end.

-file("src/gabsurd/task.gleam", 168).
-spec encode_retry_strategy(retry_strategy()) -> gleam@json:json().
encode_retry_strategy(Strategy) ->
    case Strategy of
        {fixed_retry, Base_seconds} ->
            gleam@json:object(
                [{<<"kind"/utf8>>, gleam@json:string(<<"fixed"/utf8>>)},
                    {<<"base_seconds"/utf8>>, gleam@json:int(Base_seconds)}]
            );

        {exponential_retry, Base_seconds@1, Factor, Max_seconds} ->
            Entries = [{<<"kind"/utf8>>,
                    gleam@json:string(<<"exponential"/utf8>>)},
                {<<"base_seconds"/utf8>>, gleam@json:int(Base_seconds@1)},
                {<<"factor"/utf8>>, gleam@json:float(Factor)}],
            Entries@1 = case Max_seconds of
                {some, Max} ->
                    [{<<"max_seconds"/utf8>>, gleam@json:float(Max)} | Entries];

                none ->
                    Entries
            end,
            gleam@json:object(Entries@1)
    end.

-file("src/gabsurd/task.gleam", 140).
?DOC(" Encode spawn options to a JSON string for the database.\n").
-spec encode_options(spawn_options()) -> binary().
encode_options(Options) ->
    Entries = [],
    Entries@1 = case erlang:element(2, Options) of
        {some, Max} ->
            [{<<"max_attempts"/utf8>>, gleam@json:int(Max)} | Entries];

        none ->
            Entries
    end,
    Entries@2 = case erlang:element(3, Options) of
        {some, Strategy} ->
            [{<<"retry_strategy"/utf8>>, encode_retry_strategy(Strategy)} |
                Entries@1];

        none ->
            Entries@1
    end,
    Entries@3 = case erlang:element(4, Options) of
        {some, C} ->
            [{<<"cancellation"/utf8>>, encode_cancellation(C)} | Entries@2];

        none ->
            Entries@2
    end,
    Entries@4 = case erlang:element(5, Options) of
        {some, H} ->
            [{<<"headers"/utf8>>, H} | Entries@3];

        none ->
            Entries@3
    end,
    Entries@5 = case erlang:element(6, Options) of
        {some, Key} ->
            [{<<"idempotency_key"/utf8>>, gleam@json:string(Key)} | Entries@4];

        none ->
            Entries@4
    end,
    gleam@json:to_string(gleam@json:object(Entries@5)).

-file("src/gabsurd/task.gleam", 219).
?DOC(
    " Spawn a new task in a queue with typed options.\n"
    "\n"
    " `params` is a `json.Json` value — use `json.object`, `json.string`, etc.\n"
    " to build it. `options` is a `SpawnOptions` record — use `new_options()` and\n"
    " `with_*` builders.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " let assert Ok(info) = task.spawn(\n"
    "   db,\n"
    "   \"emails\",\n"
    "   \"send_welcome\",\n"
    "   json.object([#(\"to\", json.string(\"user@example.com\"))]),\n"
    "   task.new_options() |> task.with_max_attempts(3),\n"
    " )\n"
    " ```\n"
).
-spec spawn(
    gabsurd@client:db(),
    binary(),
    binary(),
    gleam@json:json(),
    spawn_options()
) -> {ok, spawn_info()} | {error, gabsurd@client:gabsurd_error()}.
spawn(Db, Queue_name, Task_name, Params, Options) ->
    Params_str = gleam@json:to_string(Params),
    Options_str = encode_options(Options),
    gleam@result:'try'(
        gabsurd@client:query_one(
            Db,
            gabsurd@sql:spawn_task(
                Queue_name,
                Task_name,
                Params_str,
                Options_str
            )
        ),
        fun(Row) ->
            {ok,
                {spawn_info,
                    erlang:element(2, Row),
                    erlang:element(3, Row),
                    erlang:element(4, Row),
                    erlang:element(5, Row)}}
        end
    ).

-file("src/gabsurd/task.gleam", 243).
?DOC(" Claim available tasks from a queue for a worker.\n").
-spec claim(gabsurd@client:db(), binary(), binary(), integer(), integer()) -> {ok,
        list(claim())} |
    {error, gabsurd@client:gabsurd_error()}.
claim(Db, Queue_name, Worker_id, Claim_timeout, Qty) ->
    gleam@result:'try'(
        gabsurd@client:query_many(
            Db,
            gabsurd@sql:claim_task(Queue_name, Worker_id, Claim_timeout, Qty)
        ),
        fun(Rows) ->
            {ok,
                gleam@list:map(
                    Rows,
                    fun(Row) ->
                        {claim,
                            erlang:element(2, Row),
                            erlang:element(3, Row),
                            erlang:element(4, Row),
                            erlang:element(5, Row),
                            erlang:element(6, Row),
                            erlang:element(7, Row),
                            erlang:element(8, Row),
                            erlang:element(9, Row),
                            erlang:element(10, Row),
                            erlang:element(11, Row)}
                    end
                )}
        end
    ).

-file("src/gabsurd/task.gleam", 275).
?DOC(" Mark a run as completed with optional result state.\n").
-spec complete(gabsurd@client:db(), binary(), bitstring(), gleam@json:json()) -> {ok,
        nil} |
    {error, gabsurd@client:gabsurd_error()}.
complete(Db, Queue_name, Run_id, State) ->
    gabsurd@client:exec(
        Db,
        gabsurd@sql:complete_run(
            Queue_name,
            Run_id,
            gleam@json:to_string(State)
        )
    ).

-file("src/gabsurd/task.gleam", 289).
?DOC(
    " Mark a run as failed with a reason.\n"
    " Passes NULL for retry_at so the queue's retry policy controls retries.\n"
).
-spec fail(gabsurd@client:db(), binary(), bitstring(), gleam@json:json()) -> {ok,
        nil} |
    {error, gabsurd@client:gabsurd_error()}.
fail(Db, Queue_name, Run_id, Reason) ->
    gabsurd@client:exec(
        Db,
        gabsurd@sql:fail_run(Queue_name, Run_id, gleam@json:to_string(Reason))
    ).

-file("src/gabsurd/task.gleam", 302).
?DOC(" Mark a run as failed and schedule a retry at a specific time.\n").
-spec fail_with_retry(
    gabsurd@client:db(),
    binary(),
    bitstring(),
    gleam@json:json(),
    gleam@time@timestamp:timestamp()
) -> {ok, nil} | {error, gabsurd@client:gabsurd_error()}.
fail_with_retry(Db, Queue_name, Run_id, Reason, Retry_at) ->
    gabsurd@client:exec(
        Db,
        gabsurd@sql:fail_run_with_retry(
            Queue_name,
            Run_id,
            gleam@json:to_string(Reason),
            Retry_at
        )
    ).

-file("src/gabsurd/task.gleam", 328).
?DOC(
    " Extend a worker's claim lease on a run by `extend_by` seconds.\n"
    "\n"
    " This is the manual heartbeat mechanism. The primary lease extension\n"
    " mechanism is `checkpoint.set` which calls `set_task_checkpoint_state`\n"
    " with `extend_claim_by` — every checkpoint write extends the lease.\n"
    "\n"
    " Use this function when you have long-running work between checkpoints\n"
    " and need to keep the lease alive.\n"
).
-spec extend_claim(gabsurd@client:db(), binary(), bitstring(), integer()) -> {ok,
        nil} |
    {error, gabsurd@client:gabsurd_error()}.
extend_claim(Db, Queue_name, Run_id, Extend_by) ->
    gabsurd@client:exec(
        Db,
        gabsurd@sql:extend_claim(Queue_name, Run_id, Extend_by)
    ).

-file("src/gabsurd/task.gleam", 341).
?DOC(
    " Schedule a run to become available again at a future time.\n"
    " Used for deferring unknown tasks during rolling deployments.\n"
    "\n"
    " `defer_seconds` is how many seconds from now to reschedule.\n"
).
-spec schedule_run(gabsurd@client:db(), binary(), bitstring(), integer()) -> {ok,
        nil} |
    {error, gabsurd@client:gabsurd_error()}.
schedule_run(Db, Queue_name, Run_id, Defer_seconds) ->
    Now = gleam@time@timestamp:system_time(),
    Wake_at = gleam@time@timestamp:add(
        Now,
        gleam@time@duration:seconds(Defer_seconds)
    ),
    gabsurd@client:exec(
        Db,
        gabsurd@sql:schedule_run(Queue_name, Run_id, Wake_at)
    ).

-file("src/gabsurd/task.gleam", 353).
?DOC(" Cancel a task by its task_id.\n").
-spec cancel(gabsurd@client:db(), binary(), bitstring()) -> {ok, nil} |
    {error, gabsurd@client:gabsurd_error()}.
cancel(Db, Queue_name, Task_id) ->
    gabsurd@client:exec(Db, gabsurd@sql:cancel_task(Queue_name, Task_id)).

-file("src/gabsurd/task.gleam", 362).
?DOC(" Get the result of a completed task.\n").
-spec get_result(gabsurd@client:db(), binary(), bitstring()) -> {ok,
        task_result()} |
    {error, gabsurd@client:gabsurd_error()}.
get_result(Db, Queue_name, Task_id) ->
    gleam@result:'try'(
        gabsurd@client:query_one(
            Db,
            gabsurd@sql:get_task_result(Queue_name, Task_id)
        ),
        fun(Row) ->
            {ok,
                {task_result,
                    erlang:element(3, Row),
                    erlang:element(4, Row),
                    erlang:element(5, Row)}}
        end
    ).

-file("src/gabsurd/task.gleam", 381).
?DOC(" Retry a task with typed options.\n").
-spec retry(gabsurd@client:db(), binary(), bitstring(), spawn_options()) -> {ok,
        spawn_info()} |
    {error, gabsurd@client:gabsurd_error()}.
retry(Db, Queue_name, Task_id, Options) ->
    Options_str = encode_options(Options),
    gleam@result:'try'(
        gabsurd@client:query_one(
            Db,
            gabsurd@sql:retry_task(Queue_name, Task_id, Options_str)
        ),
        fun(Row) ->
            {ok,
                {spawn_info,
                    erlang:element(2, Row),
                    erlang:element(3, Row),
                    erlang:element(4, Row),
                    erlang:element(5, Row)}}
        end
    ).
Operator	Description	Example
name:	Match package name. Supports * wildcards and repo/package form	name:phx* or name:hexpm/phoenix
description:	Full-text search of package descriptions	description:auth
depends:	Packages depending on a given package. Supports repo:package form	depends:ecto or depends:hexpm:ecto
build_tool:	Filter by build tool	build_tool:mix
updated_after:	Packages updated after an ISO8601 datetime	updated_after:2025-01-01T00:00:00Z
extra:	Match custom metadata (key,value). Nested keys are separated by commas	extra:license,MIT