Current section
Files
Jump to
Current section
Files
src/m25@internal@sql_ext.erl
-module(m25@internal@sql_ext).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).
-define(FILEPATH, "src/m25/internal/sql_ext.gleam").
-export([insert_job/12, reserve_jobs/3, get_job/2, cancel_job/2]).
-export_type([job_record_row/0, cancel_outcome/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(false).
-type job_record_row() :: {job_record_row,
youid@uuid:uuid(),
binary(),
gleam@time@timestamp:timestamp(),
gleam@option:option(gleam@time@timestamp:timestamp()),
binary(),
gleam@option:option(gleam@time@timestamp:timestamp()),
gleam@option:option(gleam@time@timestamp:timestamp()),
gleam@option:option(gleam@time@timestamp:timestamp()),
gleam@option:option(gleam@time@timestamp:timestamp()),
integer(),
binary(),
gleam@option:option(binary()),
gleam@option:option(gleam@time@timestamp:timestamp()),
gleam@option:option(gleam@time@timestamp:timestamp()),
gleam@option:option(binary()),
gleam@option:option(binary()),
integer(),
integer(),
gleam@option:option(youid@uuid:uuid()),
gleam@option:option(youid@uuid:uuid()),
integer(),
gleam@option:option(binary())}.
-type cancel_outcome() :: cancelled | not_pending.
-file("src/m25/internal/sql_ext.gleam", 380).
?DOC(false).
-spec uuid_decoder() -> gleam@dynamic@decode:decoder(youid@uuid:uuid()).
uuid_decoder() ->
gleam@dynamic@decode:then(
{decoder, fun gleam@dynamic@decode:decode_bit_array/1},
fun(Bit_array) -> case youid@uuid:from_bit_array(Bit_array) of
{ok, Uuid} ->
gleam@dynamic@decode:success(Uuid);
{error, _} ->
gleam@dynamic@decode:failure(
youid@uuid:v7(),
<<"Uuid"/utf8>>
)
end end
).
-file("src/m25/internal/sql_ext.gleam", 40).
?DOC(false).
-spec job_record_row_decoder() -> gleam@dynamic@decode:decoder(job_record_row()).
job_record_row_decoder() ->
gleam@dynamic@decode:field(
0,
uuid_decoder(),
fun(Id) ->
gleam@dynamic@decode:field(
1,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Queue_name) ->
gleam@dynamic@decode:field(
2,
pog:timestamp_decoder(),
fun(Created_at) ->
gleam@dynamic@decode:field(
3,
gleam@dynamic@decode:optional(
pog:timestamp_decoder()
),
fun(Scheduled_at) ->
gleam@dynamic@decode:field(
4,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Input) ->
gleam@dynamic@decode:field(
5,
gleam@dynamic@decode:optional(
pog:timestamp_decoder()
),
fun(Reserved_at) ->
gleam@dynamic@decode:field(
6,
gleam@dynamic@decode:optional(
pog:timestamp_decoder(
)
),
fun(Started_at) ->
gleam@dynamic@decode:field(
7,
gleam@dynamic@decode:optional(
pog:timestamp_decoder(
)
),
fun(
Cancelled_at
) ->
gleam@dynamic@decode:field(
8,
gleam@dynamic@decode:optional(
pog:timestamp_decoder(
)
),
fun(
Finished_at
) ->
gleam@dynamic@decode:field(
9,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(
Timeout
) ->
gleam@dynamic@decode:field(
10,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(
Status
) ->
gleam@dynamic@decode:field(
11,
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_string/1}
),
fun(
Output
) ->
gleam@dynamic@decode:field(
12,
gleam@dynamic@decode:optional(
pog:timestamp_decoder(
)
),
fun(
Deadline
) ->
gleam@dynamic@decode:field(
13,
gleam@dynamic@decode:optional(
pog:timestamp_decoder(
)
),
fun(
Latest_heartbeat_at
) ->
gleam@dynamic@decode:field(
14,
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_string/1}
),
fun(
Failure_reason
) ->
gleam@dynamic@decode:field(
15,
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_string/1}
),
fun(
Error_data
) ->
gleam@dynamic@decode:field(
16,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(
Attempt
) ->
gleam@dynamic@decode:field(
17,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(
Max_attempts
) ->
gleam@dynamic@decode:field(
18,
gleam@dynamic@decode:optional(
uuid_decoder(
)
),
fun(
Original_attempt_id
) ->
gleam@dynamic@decode:field(
19,
gleam@dynamic@decode:optional(
uuid_decoder(
)
),
fun(
Previous_attempt_id
) ->
gleam@dynamic@decode:field(
20,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(
Retry_delay
) ->
gleam@dynamic@decode:field(
21,
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_string/1}
),
fun(
Unique_key
) ->
gleam@dynamic@decode:success(
{job_record_row,
Id,
Queue_name,
Created_at,
Scheduled_at,
Input,
Reserved_at,
Started_at,
Cancelled_at,
Finished_at,
Timeout,
Status,
Output,
Deadline,
Latest_heartbeat_at,
Failure_reason,
Error_data,
Attempt,
Max_attempts,
Original_attempt_id,
Previous_attempt_id,
Retry_delay,
Unique_key}
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
).
-file("src/m25/internal/sql_ext.gleam", 98).
?DOC(false).
-spec insert_job(
pog:connection(),
youid@uuid:uuid(),
binary(),
gleam@option:option(float()),
gleam@json:json(),
float(),
integer(),
integer(),
gleam@option:option(youid@uuid:uuid()),
gleam@option:option(youid@uuid:uuid()),
float(),
gleam@option:option(binary())
) -> {ok, pog:returned(job_record_row())} | {error, pog:query_error()}.
insert_job(
Db,
Arg_1,
Arg_2,
Arg_3,
Arg_4,
Arg_5,
Arg_6,
Arg_7,
Arg_8,
Arg_9,
Arg_10,
Arg_11
) ->
_pipe = <<"insert into m25.job (
id,
queue_name,
scheduled_at,
input,
timeout,
attempt,
max_attempts,
original_attempt_id,
previous_attempt_id,
retry_delay,
unique_key
) values (
$1,
$2,
to_timestamp($3),
$4,
make_interval(secs => $5),
$6,
$7,
$8,
$9,
make_interval(secs => $10),
$11
) returning
id,
queue_name,
created_at::timestamp,
scheduled_at::timestamp,
input,
reserved_at::timestamp,
started_at::timestamp,
cancelled_at::timestamp,
finished_at::timestamp,
extract(epoch from timeout)::int as timeout,
status,
output,
deadline::timestamp,
latest_heartbeat_at::timestamp,
failure_reason,
error_data,
attempt,
max_attempts,
original_attempt_id,
previous_attempt_id,
extract(epoch from retry_delay)::int as retry_delay,
unique_key;
"/utf8>>,
_pipe@1 = pog:'query'(_pipe),
_pipe@2 = pog:parameter(
_pipe@1,
pog_ffi:coerce(youid@uuid:to_string(Arg_1))
),
_pipe@3 = pog:parameter(_pipe@2, pog_ffi:coerce(Arg_2)),
_pipe@4 = pog:parameter(_pipe@3, pog:nullable(fun pog_ffi:coerce/1, Arg_3)),
_pipe@5 = pog:parameter(
_pipe@4,
pog_ffi:coerce(gleam@json:to_string(Arg_4))
),
_pipe@6 = pog:parameter(_pipe@5, pog_ffi:coerce(Arg_5)),
_pipe@7 = pog:parameter(_pipe@6, pog_ffi:coerce(Arg_6)),
_pipe@8 = pog:parameter(_pipe@7, pog_ffi:coerce(Arg_7)),
_pipe@9 = pog:parameter(
_pipe@8,
pog:nullable(
fun pog_ffi:coerce/1,
gleam@option:map(Arg_8, fun youid@uuid:to_string/1)
)
),
_pipe@10 = pog:parameter(
_pipe@9,
pog:nullable(
fun pog_ffi:coerce/1,
gleam@option:map(Arg_9, fun youid@uuid:to_string/1)
)
),
_pipe@11 = pog:parameter(_pipe@10, pog_ffi:coerce(Arg_10)),
_pipe@12 = pog:parameter(
_pipe@11,
pog:nullable(fun pog_ffi:coerce/1, Arg_11)
),
_pipe@13 = pog:returning(_pipe@12, job_record_row_decoder()),
pog:execute(_pipe@13, Db).
-file("src/m25/internal/sql_ext.gleam", 182).
?DOC(false).
-spec reserve_jobs(pog:connection(), binary(), integer()) -> {ok,
pog:returned(job_record_row())} |
{error, pog:query_error()}.
reserve_jobs(Db, Arg_1, Arg_2) ->
_pipe = <<"update m25.job
set
reserved_at = now()
where id in (
select id
from m25.job
where queue_name = $1
and status = 'pending'
and (scheduled_at <= now() or scheduled_at is null)
order by created_at
limit $2
for update skip locked
)
returning
id,
queue_name,
created_at::timestamp,
scheduled_at::timestamp,
input,
reserved_at::timestamp,
started_at::timestamp,
cancelled_at::timestamp,
finished_at::timestamp,
extract(epoch from timeout)::int as timeout,
status,
output,
deadline::timestamp,
latest_heartbeat_at::timestamp,
failure_reason,
error_data,
attempt,
max_attempts,
original_attempt_id,
previous_attempt_id,
extract(epoch from retry_delay)::int as retry_delay,
unique_key;
"/utf8>>,
_pipe@1 = pog:'query'(_pipe),
_pipe@2 = pog:parameter(_pipe@1, pog_ffi:coerce(Arg_1)),
_pipe@3 = pog:parameter(_pipe@2, pog_ffi:coerce(Arg_2)),
_pipe@4 = pog:returning(_pipe@3, job_record_row_decoder()),
pog:execute(_pipe@4, Db).
-file("src/m25/internal/sql_ext.gleam", 233).
?DOC(false).
-spec get_job(pog:connection(), youid@uuid:uuid()) -> {ok,
pog:returned(job_record_row())} |
{error, pog:query_error()}.
get_job(Db, Arg_1) ->
_pipe = <<"select
id,
queue_name,
created_at::timestamp,
scheduled_at::timestamp,
input,
reserved_at::timestamp,
started_at::timestamp,
cancelled_at::timestamp,
finished_at::timestamp,
extract(epoch from timeout)::int as timeout,
status,
output,
deadline::timestamp,
latest_heartbeat_at::timestamp,
failure_reason,
error_data,
attempt,
max_attempts,
original_attempt_id,
previous_attempt_id,
extract(epoch from retry_delay)::int as retry_delay,
unique_key
from m25.job where id = $1;
"/utf8>>,
_pipe@1 = pog:'query'(_pipe),
_pipe@2 = pog:parameter(
_pipe@1,
pog_ffi:coerce(youid@uuid:to_string(Arg_1))
),
_pipe@3 = pog:returning(_pipe@2, job_record_row_decoder()),
pog:execute(_pipe@3, Db).
-file("src/m25/internal/sql_ext.gleam", 275).
?DOC(false).
-spec cancel_job(pog:connection(), youid@uuid:uuid()) -> {ok,
pog:returned({job_record_row(), cancel_outcome()})} |
{error, pog:query_error()}.
cancel_job(Db, Arg_1) ->
Decoder = begin
gleam@dynamic@decode:then(
job_record_row_decoder(),
fun(Job_record) ->
gleam@dynamic@decode:field(
22,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Cancel_outcome) -> case Cancel_outcome of
<<"cancelled"/utf8>> ->
gleam@dynamic@decode:success(
{Job_record, cancelled}
);
<<"not_pending"/utf8>> ->
gleam@dynamic@decode:success(
{Job_record, not_pending}
);
_ ->
gleam@dynamic@decode:failure(
{Job_record, not_pending},
<<"Cancel Outcome"/utf8>>
)
end end
)
end
)
end,
_pipe = <<"-- Cancel a job only if it's pending, and distinguish outcomes in one query
with target as (
select
id,
queue_name,
created_at::timestamp,
scheduled_at::timestamp,
input,
reserved_at::timestamp,
started_at::timestamp,
cancelled_at::timestamp,
finished_at::timestamp,
extract(epoch from timeout)::int as timeout,
status,
output,
deadline::timestamp,
latest_heartbeat_at::timestamp,
failure_reason,
error_data,
attempt,
max_attempts,
original_attempt_id,
previous_attempt_id,
extract(epoch from retry_delay)::int as retry_delay,
unique_key
from m25.job where id = $1
),
updated as (
update m25.job j
set cancelled_at = now()
from target t
where j.id = t.id
and t.status = 'pending'
returning
j.id,
j.queue_name,
j.created_at::timestamp,
j.scheduled_at::timestamp,
j.input,
j.reserved_at::timestamp,
j.started_at::timestamp,
j.cancelled_at::timestamp,
j.finished_at::timestamp,
extract(epoch from j.timeout)::int as timeout,
j.status,
j.output,
j.deadline::timestamp,
j.latest_heartbeat_at::timestamp,
j.failure_reason,
j.error_data,
j.attempt,
j.max_attempts,
j.original_attempt_id,
j.previous_attempt_id,
extract(epoch from j.retry_delay)::int as retry_delay,
j.unique_key
)
select
coalesce(u.id, t.id) as id,
coalesce(u.queue_name, t.queue_name) as queue_name,
coalesce(u.created_at, t.created_at)::timestamp as created_at,
coalesce(u.scheduled_at, t.scheduled_at)::timestamp as scheduled_at,
coalesce(u.input, t.input) as input,
coalesce(u.reserved_at, t.reserved_at)::timestamp as reserved_at,
coalesce(u.started_at, t.started_at)::timestamp as started_at,
coalesce(u.cancelled_at, t.cancelled_at)::timestamp as cancelled_at,
coalesce(u.finished_at, t.finished_at)::timestamp as finished_at,
coalesce(u.timeout, t.timeout)::int as timeout,
coalesce(u.status, t.status) as status,
coalesce(u.output, t.output) as output,
coalesce(u.deadline, t.deadline)::timestamp as deadline,
coalesce(u.latest_heartbeat_at, t.latest_heartbeat_at)::timestamp as latest_heartbeat_at,
coalesce(u.failure_reason, t.failure_reason) as failure_reason,
coalesce(u.error_data, t.error_data) as error_data,
coalesce(u.attempt, t.attempt) as attempt,
coalesce(u.max_attempts, t.max_attempts) as max_attempts,
coalesce(u.original_attempt_id, t.original_attempt_id) as original_attempt_id,
coalesce(u.previous_attempt_id, t.previous_attempt_id) as previous_attempt_id,
coalesce(u.retry_delay, t.retry_delay) as retry_delay,
coalesce(u.unique_key, t.unique_key) as unique_key,
case when u.id is not null then 'cancelled' else 'not_pending' end as cancel_outcome
from target t
left join updated u on u.id = t.id;
"/utf8>>,
_pipe@1 = pog:'query'(_pipe),
_pipe@2 = pog:parameter(
_pipe@1,
pog_ffi:coerce(youid@uuid:to_string(Arg_1))
),
_pipe@3 = pog:returning(_pipe@2, Decoder),
pog:execute(_pipe@3, Db).